cmd/

frontend.rs

// Copyright 2023 Greptime Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

use std::fmt::Debug;
use std::future::Future;
use std::path::Path;
use std::pin::Pin;
use std::sync::Arc;
use std::time::Duration;

use async_trait::async_trait;
use cache::{build_fundamental_cache_registry, with_default_composite_cache_registry};
use catalog::information_extension::DistributedInformationExtension;
use catalog::information_schema::InformationExtensionRef;
use catalog::kvbackend::{
    CachedKvBackendBuilder, CatalogManagerConfiguratorRef, KvBackendCatalogManagerBuilder,
    new_read_only_meta_kv_backend,
};
use catalog::process_manager::ProcessManager;
use clap::Parser;
use client::client_manager::NodeClients;
use common_base::Plugins;
use common_config::{Configurable, DEFAULT_DATA_HOME};
use common_error::ext::BoxedError;
use common_meta::cache::{CacheRegistryBuilder, LayeredCacheRegistryBuilder};
use common_query::prelude::set_default_prefix;
use common_stat::ResourceStatImpl;
use common_telemetry::info;
use common_telemetry::logging::{DEFAULT_LOGGING_DIR, TracingOptions};
use common_time::timezone::set_default_timezone;
use common_version::{short_version, verbose_version};
use frontend::frontend::Frontend;
use frontend::heartbeat::{
    FrontendHeartbeatExtensions, HeartbeatTask, heartbeat_response_handler_executor,
};
use frontend::instance::builder::FrontendBuilder;
use frontend::server::Services;
use meta_client::{MetaClientOptions, MetaClientRef, MetaClientType};
use plugins::PluginOptions;
use plugins::frontend::context::{
    CatalogManagerConfigureContext, DistributedCatalogManagerConfigureContext,
};
use plugins::options::PluginOptionsDeserializerImpl;
use servers::addrs;
use servers::grpc::GrpcOptions;
use servers::tls::{TlsMode, TlsOption, merge_tls_option};
use snafu::{OptionExt, ResultExt};
use tracing_appender::non_blocking::WorkerGuard;

use crate::error::{self, OtherSnafu, Result};
use crate::options::{GlobalOptions, GreptimeOptions};
use crate::{App, create_resource_limit_metrics, log_versions, maybe_activate_heap_profile};

type FrontendOptions = GreptimeOptions<frontend::frontend::FrontendOptions>;
type HeartbeatExtensionSetupFuture<'a> = Pin<Box<dyn Future<Output = Result<()>> + Send + 'a>>;

pub struct Instance {
    frontend: Frontend,
    // Keep the logging guard to prevent the worker from being dropped.
    _guard: Vec<WorkerGuard>,
}

pub const APP_NAME: &str = "greptime-frontend";

impl Instance {
    pub fn new(frontend: Frontend, _guard: Vec<WorkerGuard>) -> Self {
        Self { frontend, _guard }
    }

    pub fn inner(&self) -> &Frontend {
        &self.frontend
    }

    pub fn mut_inner(&mut self) -> &mut Frontend {
        &mut self.frontend
    }
}

#[async_trait]
impl App for Instance {
    fn name(&self) -> &str {
        APP_NAME
    }

    async fn start(&mut self) -> Result<()> {
        plugins::start_frontend_plugins(&self.frontend.instance)
            .await
            .context(error::StartFrontendSnafu)?;

        self.frontend
            .start()
            .await
            .context(error::StartFrontendSnafu)
    }

    async fn stop(&mut self) -> Result<()> {
        self.frontend
            .shutdown()
            .await
            .context(error::ShutdownFrontendSnafu)
    }
}

#[derive(Parser)]
pub struct Command {
    #[clap(subcommand)]
    pub subcmd: SubCommand,
}

impl Command {
    pub async fn build(&self, opts: FrontendOptions) -> Result<Instance> {
        self.build_with_heartbeat_extensions(opts, FrontendHeartbeatExtensions::default())
            .await
    }

    /// Builds a frontend with pre-registered heartbeat extensions.
    ///
    /// Register extensions before calling this method. The supplied registry is installed before
    /// normal plugin setup. After plugin heartbeat setup completes, the registry is frozen before
    /// heartbeat handlers and the heartbeat task are built, so later registrations are rejected
    /// and every heartbeat consumer observes the same extension membership.
    pub async fn build_with_heartbeat_extensions(
        &self,
        opts: FrontendOptions,
        heartbeat_extensions: FrontendHeartbeatExtensions,
    ) -> Result<Instance> {
        let plugins = plugins_with_heartbeat_extensions(heartbeat_extensions);
        self.forward_build_with_plugins(opts, plugins, |command, opts, plugins| {
            command.build_with_plugins(opts, plugins)
        })
        .await
    }

    fn forward_build_with_plugins<'a, T>(
        &'a self,
        opts: FrontendOptions,
        plugins: Plugins,
        build: impl FnOnce(&'a StartCommand, FrontendOptions, Plugins) -> T,
    ) -> T {
        self.subcmd.forward_build_with_plugins(opts, plugins, build)
    }

    pub fn load_options(&self, global_options: &GlobalOptions) -> Result<FrontendOptions> {
        self.subcmd.load_options(global_options)
    }
}

#[derive(Parser)]
pub enum SubCommand {
    Start(StartCommand),
}

impl SubCommand {
    fn forward_build_with_plugins<'a, T>(
        &'a self,
        opts: FrontendOptions,
        plugins: Plugins,
        build: impl FnOnce(&'a StartCommand, FrontendOptions, Plugins) -> T,
    ) -> T {
        match self {
            SubCommand::Start(cmd) => build(cmd, opts, plugins),
        }
    }

    fn load_options(&self, global_options: &GlobalOptions) -> Result<FrontendOptions> {
        match self {
            SubCommand::Start(cmd) => cmd.load_options(global_options),
        }
    }
}

#[derive(Debug, Default, Parser)]
pub struct StartCommand {
    /// The address to bind the gRPC server.
    #[clap(long = "grpc-bind-addr", alias = "rpc-bind-addr", alias = "rpc-addr")]
    grpc_bind_addr: Option<String>,
    /// The address advertised to the metasrv, and used for connections from outside the host.
    /// If left empty or unset, the server will automatically use the IP address of the first network interface
    /// on the host, with the same port number as the one specified in `grpc_bind_addr`.
    #[clap(
        long = "grpc-server-addr",
        alias = "rpc-server-addr",
        alias = "rpc-hostname"
    )]
    grpc_server_addr: Option<String>,
    /// The address to bind the internal gRPC server.
    #[clap(
        long = "internal-grpc-bind-addr",
        alias = "internal-rpc-bind-addr",
        alias = "internal-rpc-addr"
    )]
    internal_grpc_bind_addr: Option<String>,
    /// The address advertised to the metasrv, and used for connections from outside the host.
    /// If left empty or unset, the server will automatically use the IP address of the first network interface
    /// on the host, with the same port number as the one specified in `internal_grpc_bind_addr`.
    #[clap(
        long = "internal-grpc-server-addr",
        alias = "internal-rpc-server-addr",
        alias = "internal-rpc-hostname"
    )]
    internal_grpc_server_addr: Option<String>,
    #[clap(long)]
    http_addr: Option<String>,
    #[clap(long)]
    http_timeout: Option<u64>,
    #[clap(long)]
    mysql_addr: Option<String>,
    #[clap(long)]
    postgres_addr: Option<String>,
    #[clap(short, long)]
    pub config_file: Option<String>,
    #[clap(short, long)]
    influxdb_enable: Option<bool>,
    #[clap(long, value_delimiter = ',', num_args = 1..)]
    metasrv_addrs: Option<Vec<String>>,
    #[clap(long)]
    tls_mode: Option<TlsMode>,
    #[clap(long)]
    tls_cert_path: Option<String>,
    #[clap(long)]
    tls_key_path: Option<String>,
    #[clap(long)]
    tls_watch: bool,
    #[clap(long)]
    user_provider: Option<String>,
    #[clap(long)]
    disable_dashboard: Option<bool>,
    #[clap(long, default_value = "GREPTIMEDB_FRONTEND")]
    pub env_prefix: String,
}

impl StartCommand {
    fn load_options(&self, global_options: &GlobalOptions) -> Result<FrontendOptions> {
        let mut opts = FrontendOptions::load_layered_options(
            self.config_file.as_deref(),
            self.env_prefix.as_ref(),
        )
        .context(error::LoadLayeredConfigSnafu)?;

        self.merge_with_cli_options(global_options, &mut opts)?;

        Ok(opts)
    }

    // The precedence order is: cli > config file > environment variables > default values.
    fn merge_with_cli_options(
        &self,
        global_options: &GlobalOptions,
        opts: &mut FrontendOptions,
    ) -> Result<()> {
        let opts = &mut opts.component;

        if let Some(dir) = &global_options.log_dir {
            opts.logging.dir.clone_from(dir);
        }

        // If the logging dir is not set, use the default logs dir in the data home.
        if opts.logging.dir.is_empty() {
            opts.logging.dir = Path::new(DEFAULT_DATA_HOME)
                .join(DEFAULT_LOGGING_DIR)
                .to_string_lossy()
                .to_string();
        }

        if global_options.log_level.is_some() {
            opts.logging.level.clone_from(&global_options.log_level);
        }

        opts.tracing = TracingOptions {
            #[cfg(feature = "tokio-console")]
            tokio_console_addr: global_options.tokio_console_addr.clone(),
        };

        let tls_opts = TlsOption::new(
            self.tls_mode,
            self.tls_cert_path.clone(),
            self.tls_key_path.clone(),
            self.tls_watch,
        );

        if let Some(addr) = &self.http_addr {
            opts.http.addr.clone_from(addr);
        }

        if let Some(http_timeout) = self.http_timeout {
            opts.http.timeout = Duration::from_secs(http_timeout)
        }

        if let Some(disable_dashboard) = self.disable_dashboard {
            opts.http.disable_dashboard = disable_dashboard;
        }

        if let Some(addr) = &self.grpc_bind_addr {
            opts.grpc.bind_addr.clone_from(addr);
            opts.grpc.tls = merge_tls_option(&opts.grpc.tls, tls_opts.clone());
        }

        if let Some(addr) = &self.grpc_server_addr {
            opts.grpc.server_addr.clone_from(addr);
        }

        if let Some(addr) = &self.internal_grpc_bind_addr {
            if let Some(internal_grpc) = &mut opts.internal_grpc {
                internal_grpc.bind_addr = addr.clone();
            } else {
                let grpc_options = GrpcOptions {
                    bind_addr: addr.clone(),
                    ..Default::default()
                };

                opts.internal_grpc = Some(grpc_options);
            }
        }

        if let Some(addr) = &self.internal_grpc_server_addr {
            if let Some(internal_grpc) = &mut opts.internal_grpc {
                internal_grpc.server_addr = addr.clone();
            } else {
                let grpc_options = GrpcOptions {
                    server_addr: addr.clone(),
                    ..Default::default()
                };
                opts.internal_grpc = Some(grpc_options);
            }
        }

        if let Some(addr) = &self.mysql_addr {
            opts.mysql.enable = true;
            opts.mysql.addr.clone_from(addr);
            opts.mysql.tls = merge_tls_option(&opts.mysql.tls, tls_opts.clone());
        }

        if let Some(addr) = &self.postgres_addr {
            opts.postgres.enable = true;
            opts.postgres.addr.clone_from(addr);
            opts.postgres.tls = merge_tls_option(&opts.postgres.tls, tls_opts.clone());
        }

        if let Some(enable) = self.influxdb_enable {
            opts.influxdb.enable = enable;
        }

        if let Some(metasrv_addrs) = &self.metasrv_addrs {
            opts.meta_client
                .get_or_insert_with(MetaClientOptions::default)
                .metasrv_addrs
                .clone_from(metasrv_addrs);
        }

        if let Some(user_provider) = &self.user_provider {
            opts.user_provider = Some(user_provider.clone());
        }

        Ok(())
    }

    async fn build_with_plugins(
        &self,
        opts: FrontendOptions,
        plugins: Plugins,
    ) -> Result<Instance> {
        let guard = common_telemetry::init_global_logging(
            APP_NAME,
            &opts.component.logging,
            &opts.component.tracing,
            opts.component.node_id.clone(),
            Some(&opts.component.slow_query),
        );

        common_runtime::init_global_runtimes(&opts.runtime);

        crate::options::flush_dropped_plugin_warnings();
        log_versions(verbose_version(), short_version(), APP_NAME);
        maybe_activate_heap_profile(&opts.component.memory);
        create_resource_limit_metrics(APP_NAME);

        info!("Frontend start command: {:#?}", self);
        info!("Frontend options: {:#?}", opts);

        let plugin_opts = opts.plugins;
        let mut opts = opts.component;
        opts.grpc.detect_server_addr();

        set_default_timezone(opts.default_timezone.as_deref()).context(error::InitTimezoneSnafu)?;
        set_default_prefix(opts.default_column_prefix.as_deref())
            .map_err(BoxedError::new)
            .context(error::BuildCliSnafu)?;

        let meta_client_options = opts
            .meta_client
            .as_ref()
            .context(error::MissingConfigSnafu {
                msg: "'meta_client'",
            })?;

        let cache_max_capacity = meta_client_options.metadata_cache_max_capacity;
        let cache_ttl = meta_client_options.metadata_cache_ttl;
        let cache_tti = meta_client_options.metadata_cache_tti;

        let meta_config: Vec<PluginOptions> = meta_client::create_meta_client(
            MetaClientType::Frontend,
            meta_client_options,
            None,
            None,
        )
        .await
        .context(error::MetaClientInitSnafu)?
        .pull_config(PluginOptionsDeserializerImpl)
        .await
        .context(error::MetaClientInitSnafu)?;

        let mut plugins =
            prepare_frontend_plugins(plugins, &plugin_opts, &opts, Some(&meta_config)).await?;

        // now initialize the meta_client with plugins
        let meta_client = meta_client::create_meta_client(
            MetaClientType::Frontend,
            meta_client_options,
            Some(&plugins),
            None,
        )
        .await
        .context(error::MetaClientInitSnafu)?;

        let readonly_meta_backend = new_read_only_meta_kv_backend(meta_client.clone());

        // TODO(discord9): add helper function to ease the creation of cache registry&such
        let cached_meta_backend = CachedKvBackendBuilder::new(readonly_meta_backend.clone())
            .cache_max_capacity(cache_max_capacity)
            .cache_ttl(cache_ttl)
            .cache_tti(cache_tti)
            .build();
        let cached_meta_backend = Arc::new(cached_meta_backend);

        // Builds cache registry
        let layered_cache_builder = LayeredCacheRegistryBuilder::default().add_cache_registry(
            CacheRegistryBuilder::default()
                .add_cache(cached_meta_backend.clone())
                .build(),
        );
        let fundamental_cache_registry =
            build_fundamental_cache_registry(readonly_meta_backend.clone());
        let mut layered_cache_builder = with_default_composite_cache_registry(
            layered_cache_builder.add_cache_registry(fundamental_cache_registry),
        )
        .context(error::BuildCacheRegistrySnafu)?;

        if let Some(plugin_cache_builder) = plugins::frontend::configure_cache_registry(&plugins) {
            layered_cache_builder =
                layered_cache_builder.add_cache_registry(plugin_cache_builder.build());
        }

        let layered_cache_registry = Arc::new(layered_cache_builder.build());

        // frontend to datanode need not timeout.
        // Some queries are expected to take long time.
        let mut channel_config = opts.datanode.client.channel_config();
        channel_config.timeout = None;
        // Source Flight streams and sink unary responses share pooled connections.
        channel_config.http2_adaptive_window = Some(true);
        if opts.grpc.flight_compression.transport_compression() {
            channel_config.accept_compression = true;
            channel_config.send_compression = true;
        }
        let client = Arc::new(NodeClients::new(channel_config));

        let information_extension = Arc::new(DistributedInformationExtension::new(
            meta_client.clone(),
            client.clone(),
        ));
        plugins.insert::<InformationExtensionRef>(information_extension.clone());

        let process_manager = Arc::new(ProcessManager::new(
            addrs::resolve_addr(&opts.grpc.bind_addr, Some(&opts.grpc.server_addr)),
            Some(meta_client.clone()),
        ));

        let builder = KvBackendCatalogManagerBuilder::new(
            information_extension,
            cached_meta_backend.clone(),
            layered_cache_registry.clone(),
        )
        .with_process_manager(process_manager.clone());
        let builder = if let Some(configurator) =
            plugins.get::<CatalogManagerConfiguratorRef<CatalogManagerConfigureContext>>()
        {
            let ctx = DistributedCatalogManagerConfigureContext {
                meta_client: meta_client.clone(),
            };
            let ctx = CatalogManagerConfigureContext::Distributed(ctx);

            configurator
                .configure(builder, ctx)
                .await
                .context(OtherSnafu)?
        } else {
            builder
        };
        let catalog_manager = builder.build();

        let builder = FrontendBuilder::new(
            opts.clone(),
            cached_meta_backend.clone(),
            layered_cache_registry.clone(),
            catalog_manager,
            client,
            meta_client.clone(),
            process_manager,
        );

        plugins::setup_frontend_plugins_post_build(&mut plugins, &plugin_opts, &builder)
            .await
            .context(error::StartFrontendSnafu)?;

        let instance = builder
            .with_plugin(plugins.clone())
            .with_local_cache_invalidator(layered_cache_registry)
            .try_build()
            .await
            .context(error::StartFrontendSnafu)?;

        let instance = Arc::new(instance);

        let heartbeat_instance = instance.clone();
        let heartbeat_extensions =
            setup_and_freeze_frontend_heartbeat_extensions(&mut plugins, move |plugins| {
                Box::pin(async move {
                    plugins::setup_frontend_heartbeat_extensions(
                        plugins,
                        &plugin_opts,
                        &heartbeat_instance,
                    )
                    .await
                    .context(error::StartFrontendSnafu)
                })
            })
            .await?;
        let heartbeat_task = Some(create_heartbeat_task_with_extensions(
            &opts,
            meta_client,
            &instance,
            heartbeat_extensions,
        ));

        let servers = Services::new(opts, instance.clone(), plugins)
            .build()
            .context(error::StartFrontendSnafu)?;

        let frontend = Frontend {
            instance,
            servers,
            heartbeat_task,
        };

        Ok(Instance::new(frontend, guard))
    }
}

async fn prepare_frontend_plugins(
    mut plugins: Plugins,
    plugin_opts: &[PluginOptions],
    opts: &frontend::frontend::FrontendOptions,
    meta_config: Option<&[PluginOptions]>,
) -> Result<Plugins> {
    plugins::setup_frontend_plugins_pre_build(&mut plugins, plugin_opts, opts, meta_config)
        .await
        .context(error::StartFrontendSnafu)?;
    Ok(plugins)
}

fn plugins_with_heartbeat_extensions(heartbeat_extensions: FrontendHeartbeatExtensions) -> Plugins {
    let plugins = Plugins::new();
    plugins.insert(heartbeat_extensions);
    plugins
}

async fn setup_and_freeze_frontend_heartbeat_extensions(
    plugins: &mut Plugins,
    setup: impl for<'a> FnOnce(&'a mut Plugins) -> HeartbeatExtensionSetupFuture<'a>,
) -> Result<FrontendHeartbeatExtensions> {
    setup(plugins).await?;
    let extensions = plugins
        .get::<FrontendHeartbeatExtensions>()
        .unwrap_or_default();
    extensions.freeze();
    Ok(extensions)
}

pub fn create_heartbeat_task(
    options: &frontend::frontend::FrontendOptions,
    meta_client: MetaClientRef,
    instance: &frontend::instance::Instance,
) -> HeartbeatTask {
    create_heartbeat_task_with_extensions(options, meta_client, instance, Default::default())
}

fn create_heartbeat_task_with_extensions(
    options: &frontend::frontend::FrontendOptions,
    meta_client: MetaClientRef,
    instance: &frontend::instance::Instance,
    extensions: FrontendHeartbeatExtensions,
) -> HeartbeatTask {
    let executor = heartbeat_response_handler_executor(
        &extensions,
        instance.suspend_state(),
        instance.cache_invalidator().clone(),
    );

    let stat = {
        let mut stat = ResourceStatImpl::default();
        stat.start_collect_cpu_usage();
        Arc::new(stat)
    };

    HeartbeatTask::new(
        instance.frontend_peer_addr().to_string(),
        options,
        meta_client,
        executor,
        stat,
    )
    .with_extensions(extensions)
}

#[cfg(test)]
mod tests {
    use std::io::Write;
    use std::sync::Arc;
    use std::time::Duration;

    use auth::{Identity, Password, UserProviderRef};
    use clap::{CommandFactory, Parser};
    use common_base::readable_size::ReadableSize;
    use common_config::ENV_VAR_SEP;
    use common_test_util::temp_dir::create_named_temp_file;
    use servers::grpc::GrpcOptions;
    use servers::http::HttpOptions;

    use super::*;
    use crate::options::GlobalOptions;

    #[derive(Debug)]
    struct TestExtension(&'static str);

    #[async_trait]
    impl frontend::heartbeat::FrontendHeartbeatExtension for TestExtension {
        fn name(&self) -> &'static str {
            self.0
        }
    }

    #[test]
    fn test_command_forwards_heartbeat_extensions_to_start_build() {
        let command = Command {
            subcmd: SubCommand::Start(StartCommand::default()),
        };
        let extensions = FrontendHeartbeatExtensions::default();
        let shared_extensions = extensions.clone();

        let forwarded_plugins = command.forward_build_with_plugins(
            FrontendOptions::default(),
            plugins_with_heartbeat_extensions(extensions),
            |_, _, plugins| plugins,
        );
        let forwarded_extensions = forwarded_plugins
            .get::<FrontendHeartbeatExtensions>()
            .unwrap();

        assert_registry_identity(&shared_extensions, &forwarded_extensions);
    }

    #[tokio::test]
    async fn test_prefilled_heartbeat_extensions_survive_pre_build_setup() {
        let extensions = FrontendHeartbeatExtensions::default();
        let shared_extensions = extensions.clone();

        let options = frontend::frontend::FrontendOptions {
            meta_client: Some(MetaClientOptions::default()),
            ..Default::default()
        };
        let plugins = prepare_frontend_plugins(
            plugins_with_heartbeat_extensions(extensions),
            &[],
            &options,
            None,
        )
        .await
        .unwrap();

        let setup_extensions = plugins.get_or_insert(FrontendHeartbeatExtensions::default);
        assert_registry_identity(&shared_extensions, &setup_extensions);
    }

    #[tokio::test]
    async fn test_heartbeat_extension_setup_completes_before_freeze() {
        let extensions = FrontendHeartbeatExtensions::default();
        assert_eq!(
            extensions.try_register(Arc::new(TestExtension("caller"))),
            Ok(())
        );
        let mut plugins = plugins_with_heartbeat_extensions(extensions.clone());

        let finalized = setup_and_freeze_frontend_heartbeat_extensions(&mut plugins, |plugins| {
            Box::pin(async move {
                let setup_extensions = plugins.get_or_insert(FrontendHeartbeatExtensions::default);
                assert_eq!(
                    setup_extensions.try_register(Arc::new(TestExtension("plugin"))),
                    Ok(())
                );
                Ok(())
            })
        })
        .await
        .unwrap();

        assert_eq!(
            finalized
                .extensions()
                .iter()
                .map(|extension| extension.name())
                .collect::<Vec<_>>(),
            ["caller", "plugin"]
        );
        assert_eq!(
            extensions.try_register(Arc::new(TestExtension("late"))),
            Err(frontend::heartbeat::RegistrationError::Frozen)
        );
        assert_eq!(finalized.len(), 2);
    }

    fn assert_registry_identity(
        expected: &FrontendHeartbeatExtensions,
        actual: &FrontendHeartbeatExtensions,
    ) {
        let extension = Arc::new(TestExtension("cmd-test-extension"));
        assert!(expected.register(extension.clone()));
        let registered = actual.extensions();
        assert_eq!(registered.len(), 1);
        assert!(Arc::ptr_eq(
            &(extension as Arc<dyn frontend::heartbeat::FrontendHeartbeatExtension>),
            &registered[0]
        ));
    }

    #[test]
    fn test_try_from_start_command() {
        let command = StartCommand {
            http_addr: Some("127.0.0.1:1234".to_string()),
            mysql_addr: Some("127.0.0.1:5678".to_string()),
            postgres_addr: Some("127.0.0.1:5432".to_string()),
            internal_grpc_bind_addr: Some("127.0.0.1:4010".to_string()),
            internal_grpc_server_addr: Some("10.0.0.24:4010".to_string()),
            influxdb_enable: Some(false),
            disable_dashboard: Some(false),
            ..Default::default()
        };

        let opts = command.load_options(&Default::default()).unwrap().component;

        assert_eq!(opts.http.addr, "127.0.0.1:1234");
        assert_eq!(ReadableSize::mb(64), opts.http.body_limit);
        assert_eq!(opts.mysql.addr, "127.0.0.1:5678");
        assert_eq!(opts.postgres.addr, "127.0.0.1:5432");

        let internal_grpc = opts.internal_grpc.as_ref().unwrap();
        assert_eq!(internal_grpc.bind_addr, "127.0.0.1:4010");
        assert_eq!(internal_grpc.server_addr, "10.0.0.24:4010");

        let default_opts = FrontendOptions::default().component;

        assert_eq!(opts.grpc.bind_addr, default_opts.grpc.bind_addr);
        assert!(opts.mysql.enable);
        assert_eq!(opts.mysql.runtime_size, default_opts.mysql.runtime_size);
        assert!(opts.postgres.enable);
        assert_eq!(
            opts.postgres.runtime_size,
            default_opts.postgres.runtime_size
        );
        assert!(opts.opentsdb.enable);

        assert!(!opts.influxdb.enable);
    }

    #[test]
    fn test_read_from_config_file() {
        let mut file = create_named_temp_file();
        let toml_str = r#"
            [http]
            addr = "127.0.0.1:4000"
            timeout = "0s"
            body_limit = "2GB"

            [opentsdb]
            enable = false

            [logging]
            level = "debug"
            dir = "./greptimedb_data/test/logs"
        "#;
        write!(file, "{}", toml_str).unwrap();

        let command = StartCommand {
            config_file: Some(file.path().to_str().unwrap().to_string()),
            disable_dashboard: Some(false),
            ..Default::default()
        };

        let fe_opts = command.load_options(&Default::default()).unwrap().component;

        assert_eq!("127.0.0.1:4000".to_string(), fe_opts.http.addr);
        assert_eq!(Duration::from_secs(0), fe_opts.http.timeout);

        assert_eq!(ReadableSize::gb(2), fe_opts.http.body_limit);

        assert_eq!("debug", fe_opts.logging.level.as_ref().unwrap());
        assert_eq!(
            "./greptimedb_data/test/logs".to_string(),
            fe_opts.logging.dir
        );
        assert!(!fe_opts.opentsdb.enable);
    }

    #[tokio::test]
    async fn test_try_from_start_command_to_anymap() {
        let fe_opts = frontend::frontend::FrontendOptions {
            http: HttpOptions {
                disable_dashboard: false,
                ..Default::default()
            },
            meta_client: Some(MetaClientOptions::default()),
            user_provider: Some("static_user_provider:cmd:test=test".to_string()),
            ..Default::default()
        };

        let mut plugins = Plugins::new();
        plugins::setup_frontend_plugins_pre_build(&mut plugins, &[], &fe_opts, None)
            .await
            .unwrap();

        let provider = plugins.get::<UserProviderRef>().unwrap();
        let result = provider
            .authenticate(
                Identity::UserId("test", None),
                Password::PlainText("test".to_string().into()),
            )
            .await;
        let _ = result.unwrap();
    }

    #[test]
    fn test_load_log_options_from_cli() {
        let cmd = StartCommand {
            disable_dashboard: Some(false),
            ..Default::default()
        };

        let options = cmd
            .load_options(&GlobalOptions {
                log_dir: Some("./greptimedb_data/test/logs".to_string()),
                log_level: Some("debug".to_string()),

                #[cfg(feature = "tokio-console")]
                tokio_console_addr: None,
            })
            .unwrap()
            .component;

        let logging_opt = options.logging;
        assert_eq!("./greptimedb_data/test/logs", logging_opt.dir);
        assert_eq!("debug", logging_opt.level.as_ref().unwrap());
    }

    #[test]
    fn test_config_precedence_order() {
        let mut file = create_named_temp_file();
        let toml_str = r#"
            [http]
            addr = "127.0.0.1:4000"

            [meta_client]
            timeout = "3s"
            connect_timeout = "5s"
            tcp_nodelay = true

            [mysql]
            addr = "127.0.0.1:4002"
        "#;
        write!(file, "{}", toml_str).unwrap();

        let env_prefix = "FRONTEND_UT";
        temp_env::with_vars(
            [
                (
                    // mysql.addr = 127.0.0.1:14002
                    [
                        env_prefix.to_string(),
                        "mysql".to_uppercase(),
                        "addr".to_uppercase(),
                    ]
                    .join(ENV_VAR_SEP),
                    Some("127.0.0.1:14002"),
                ),
                (
                    // mysql.runtime_size = 11
                    [
                        env_prefix.to_string(),
                        "mysql".to_uppercase(),
                        "runtime_size".to_uppercase(),
                    ]
                    .join(ENV_VAR_SEP),
                    Some("11"),
                ),
                (
                    // http.addr = 127.0.0.1:24000
                    [
                        env_prefix.to_string(),
                        "http".to_uppercase(),
                        "addr".to_uppercase(),
                    ]
                    .join(ENV_VAR_SEP),
                    Some("127.0.0.1:24000"),
                ),
                (
                    // meta_client.metasrv_addrs = 127.0.0.1:3001,127.0.0.1:3002,127.0.0.1:3003
                    [
                        env_prefix.to_string(),
                        "meta_client".to_uppercase(),
                        "metasrv_addrs".to_uppercase(),
                    ]
                    .join(ENV_VAR_SEP),
                    Some("127.0.0.1:3001,127.0.0.1:3002,127.0.0.1:3003"),
                ),
            ],
            || {
                let command = StartCommand {
                    config_file: Some(file.path().to_str().unwrap().to_string()),
                    http_addr: Some("127.0.0.1:14000".to_string()),
                    env_prefix: env_prefix.to_string(),
                    ..Default::default()
                };

                let fe_opts = command.load_options(&Default::default()).unwrap().component;

                // Should be read from env, env > default values.
                assert_eq!(fe_opts.mysql.runtime_size, 11);
                assert_eq!(
                    fe_opts.meta_client.unwrap().metasrv_addrs,
                    vec![
                        "127.0.0.1:3001".to_string(),
                        "127.0.0.1:3002".to_string(),
                        "127.0.0.1:3003".to_string()
                    ]
                );

                // Should be read from config file, config file > env > default values.
                assert_eq!(fe_opts.mysql.addr, "127.0.0.1:4002");

                // Should be read from cli, cli > config file > env > default values.
                assert_eq!(fe_opts.http.addr, "127.0.0.1:14000");

                // Should be default value.
                assert_eq!(fe_opts.grpc.bind_addr, GrpcOptions::default().bind_addr);
            },
        );
    }

    #[test]
    fn test_parse_grpc_cli_aliases() {
        let command = StartCommand::try_parse_from([
            "frontend",
            "--grpc-bind-addr",
            "127.0.0.1:14001",
            "--grpc-server-addr",
            "10.0.0.1:14001",
            "--internal-grpc-bind-addr",
            "127.0.0.1:14010",
            "--internal-grpc-server-addr",
            "10.0.0.1:14010",
        ])
        .unwrap();
        assert_eq!(command.grpc_bind_addr.as_deref(), Some("127.0.0.1:14001"));
        assert_eq!(command.grpc_server_addr.as_deref(), Some("10.0.0.1:14001"));
        assert_eq!(
            command.internal_grpc_bind_addr.as_deref(),
            Some("127.0.0.1:14010")
        );
        assert_eq!(
            command.internal_grpc_server_addr.as_deref(),
            Some("10.0.0.1:14010")
        );

        let command = StartCommand::try_parse_from([
            "frontend",
            "--rpc-bind-addr",
            "127.0.0.1:24001",
            "--rpc-server-addr",
            "10.0.0.2:24001",
            "--internal-rpc-bind-addr",
            "127.0.0.1:24010",
            "--internal-rpc-server-addr",
            "10.0.0.2:24010",
        ])
        .unwrap();
        assert_eq!(command.grpc_bind_addr.as_deref(), Some("127.0.0.1:24001"));
        assert_eq!(command.grpc_server_addr.as_deref(), Some("10.0.0.2:24001"));
        assert_eq!(
            command.internal_grpc_bind_addr.as_deref(),
            Some("127.0.0.1:24010")
        );
        assert_eq!(
            command.internal_grpc_server_addr.as_deref(),
            Some("10.0.0.2:24010")
        );

        let command = StartCommand::try_parse_from([
            "frontend",
            "--rpc-addr",
            "127.0.0.1:34001",
            "--rpc-hostname",
            "10.0.0.3:34001",
            "--internal-rpc-addr",
            "127.0.0.1:34010",
            "--internal-rpc-hostname",
            "10.0.0.3:34010",
        ])
        .unwrap();
        assert_eq!(command.grpc_bind_addr.as_deref(), Some("127.0.0.1:34001"));
        assert_eq!(command.grpc_server_addr.as_deref(), Some("10.0.0.3:34001"));
        assert_eq!(
            command.internal_grpc_bind_addr.as_deref(),
            Some("127.0.0.1:34010")
        );
        assert_eq!(
            command.internal_grpc_server_addr.as_deref(),
            Some("10.0.0.3:34010")
        );
    }

    #[test]
    fn test_help_uses_grpc_option_names() {
        let mut cmd = StartCommand::command();
        let mut help = Vec::new();
        cmd.write_long_help(&mut help).unwrap();
        let help = String::from_utf8(help).unwrap();

        assert!(help.contains("--grpc-bind-addr"));
        assert!(help.contains("--grpc-server-addr"));
        assert!(help.contains("--internal-grpc-bind-addr"));
        assert!(help.contains("--internal-grpc-server-addr"));
        assert!(!help.contains("--rpc-bind-addr"));
        assert!(!help.contains("--rpc-server-addr"));
        assert!(!help.contains("--rpc-addr"));
        assert!(!help.contains("--rpc-hostname"));
        assert!(!help.contains("--internal-rpc-bind-addr"));
        assert!(!help.contains("--internal-rpc-server-addr"));
        assert!(!help.contains("--internal-rpc-addr"));
        assert!(!help.contains("--internal-rpc-hostname"));
    }
}