cli.rs

#![allow(missing_docs)]

use std::{
    num::{NonZeroU64, NonZeroUsize},
    path::PathBuf,
};

use clap::{ArgAction, CommandFactory, FromArgMatches, Parser};

#[cfg(windows)]
use crate::service;
#[cfg(feature = "api-client")]
use crate::tap;
#[cfg(feature = "top")]
use crate::top;

use crate::{
    completion, config, convert_config, generate, generate_schema, get_version, graph, list,
    signal, unit_test, validate,
};

#[derive(Parser, Debug)]
#[command(rename_all = "kebab-case")]
pub struct Opts {
    #[command(flatten)]
    pub root: RootOpts,

    #[command(subcommand)]
    pub sub_command: Option<SubCommand>,
}

impl Opts {
    pub fn get_matches() -> Result<Self, clap::Error> {
        let version = get_version();
        let app = Opts::command().version(version);
        Opts::from_arg_matches(&app.get_matches())
    }

    pub const fn log_level(&self) -> &'static str {
        let (quiet_level, verbose_level) = match self.sub_command {
            Some(SubCommand::Validate(_))
            | Some(SubCommand::Graph(_))
            | Some(SubCommand::Generate(_))
            | Some(SubCommand::ConvertConfig(_))
            | Some(SubCommand::List(_))
            | Some(SubCommand::Test(_)) => {
                if self.root.verbose == 0 {
                    (self.root.quiet + 1, self.root.verbose)
                } else {
                    (self.root.quiet, self.root.verbose - 1)
                }
            }
            _ => (self.root.quiet, self.root.verbose),
        };
        match quiet_level {
            0 => match verbose_level {
                0 => "info",
                1 => "debug",
                2..=255 => "trace",
            },
            1 => "warn",
            2 => "error",
            3..=255 => "off",
        }
    }
}

#[derive(Parser, Debug)]
#[command(rename_all = "kebab-case")]
pub struct RootOpts {
    /// Read configuration from one or more files. Wildcard paths are supported.
    /// File format is detected from the file name.
    /// If zero files are specified, the deprecated default config path
    /// `/etc/vector/vector.yaml` is targeted.
    #[arg(
        id = "config",
        short,
        long,
        env = "VECTOR_CONFIG",
        value_delimiter(',')
    )]
    pub config_paths: Vec<PathBuf>,

    /// Read configuration from files in one or more directories.
    /// File format is detected from the file name.
    ///
    /// Files not ending in .toml, .json, .yaml, or .yml will be ignored.
    #[arg(
        id = "config-dir",
        short = 'C',
        long,
        env = "VECTOR_CONFIG_DIR",
        value_delimiter(',')
    )]
    pub config_dirs: Vec<PathBuf>,

    /// Read configuration from one or more files. Wildcard paths are supported.
    /// TOML file format is expected.
    #[arg(
        id = "config-toml",
        long,
        env = "VECTOR_CONFIG_TOML",
        value_delimiter(',')
    )]
    pub config_paths_toml: Vec<PathBuf>,

    /// Read configuration from one or more files. Wildcard paths are supported.
    /// JSON file format is expected.
    #[arg(
        id = "config-json",
        long,
        env = "VECTOR_CONFIG_JSON",
        value_delimiter(',')
    )]
    pub config_paths_json: Vec<PathBuf>,

    /// Read configuration from one or more files. Wildcard paths are supported.
    /// YAML file format is expected.
    #[arg(
        id = "config-yaml",
        long,
        env = "VECTOR_CONFIG_YAML",
        value_delimiter(',')
    )]
    pub config_paths_yaml: Vec<PathBuf>,

    /// Exit on startup if any sinks fail healthchecks
    #[arg(short, long, env = "VECTOR_REQUIRE_HEALTHY")]
    pub require_healthy: Option<bool>,

    /// Number of threads to use for processing (default is number of available cores)
    #[arg(short, long, env = "VECTOR_THREADS")]
    pub threads: Option<usize>,

    /// Number of events batched per source send and used as the base for source output buffer sizing
    /// (source output buffer capacity is this value multiplied by the number of worker threads)
    #[arg(long, env = "VECTOR_CHUNK_SIZE_EVENTS")]
    pub chunk_size_events: Option<NonZeroUsize>,

    /// Enable more detailed internal logging. Repeat to increase level. Overridden by `--quiet`.
    #[arg(short, long, action = ArgAction::Count)]
    pub verbose: u8,

    /// Reduce detail of internal logging. Repeat to reduce further. Overrides `--verbose`.
    #[arg(short, long, action = ArgAction::Count)]
    pub quiet: u8,

    /// Allow interpolation of environment variables in configuration files. Enabling this may
    /// expose environment secrets into your Vector configuration.
    #[arg(
        long,
        env = "VECTOR_DANGEROUSLY_ALLOW_ENV_VAR_INTERPOLATION",
        default_value = "false"
    )]
    pub dangerously_allow_env_var_interpolation: bool,

    /// Set the logging format
    #[arg(long, default_value = "text", env = "VECTOR_LOG_FORMAT")]
    pub log_format: LogFormat,

    /// Control when ANSI terminal formatting is used.
    ///
    /// By default `vector` will try and detect if `stdout` is a terminal, if it is
    /// ANSI will be enabled. Otherwise it will be disabled. By providing this flag with
    /// the `--color always` option will always enable ANSI terminal formatting. `--color never`
    /// will disable all ANSI terminal formatting. `--color auto` will attempt
    /// to detect it automatically.
    #[arg(long, default_value = "auto", env = "VECTOR_COLOR")]
    pub color: Color,

    /// Watch for changes in configuration file, and reload accordingly.
    #[arg(short, long, env = "VECTOR_WATCH_CONFIG")]
    pub watch_config: bool,

    /// Method for configuration watching.
    ///
    /// By default, `vector` uses recommended watcher for host OS
    /// - `inotify` for Linux-based systems.
    /// - `kqueue` for unix/macos
    /// - `ReadDirectoryChangesWatcher` for windows
    ///
    /// The `poll` watcher can be used in cases where `inotify` doesn't work, e.g., when attaching the configuration via NFS.
    #[arg(
        long,
        default_value = "recommended",
        env = "VECTOR_WATCH_CONFIG_METHOD"
    )]
    pub watch_config_method: WatchConfigMethod,

    /// Poll for changes in the configuration file at the given interval.
    ///
    /// This setting is only applicable if `Poll` is set in `--watch-config-method`.
    #[arg(
        long,
        env = "VECTOR_WATCH_CONFIG_POLL_INTERVAL_SECONDS",
        default_value = "30"
    )]
    pub watch_config_poll_interval_seconds: NonZeroU64,

    /// Set the internal log rate limit in seconds.
    ///
    /// This controls the time window for rate limiting Vector's own internal logs.
    /// Within each time window, the first occurrence of a log is emitted, the second
    /// shows a suppression warning, and subsequent occurrences are silent until the
    /// window expires. When the window expires and the log fires again, a summary of
    /// the suppressed count is emitted followed by the log itself.
    ///
    /// Logs are grouped by their location in the code and the `component_id` field, so logs
    /// from different components are rate limited independently.
    ///
    /// Examples:
    /// - 1: Very verbose, logs can repeat every second
    /// - 10 (default): Logs can repeat every 10 seconds
    /// - 60: Less verbose, logs can repeat every minute
    #[arg(
        short,
        long,
        env = "VECTOR_INTERNAL_LOG_RATE_LIMIT",
        default_value = "10"
    )]
    pub internal_log_rate_limit: u64,

    /// Apply a rate limit (in seconds) to the broadcast channel that feeds all `internal_logs`
    /// sources. When set, the first occurrence of a repeated log is emitted, the second shows a
    /// suppression warning, and subsequent occurrences are silent until the window expires. When
    /// the window expires and the log fires again, a summary of the suppressed count is emitted
    /// followed by the log itself. Unset by default so that `internal_logs` consumers receive
    /// every log event. This limit is independent of `--internal-log-rate-limit`, which only
    /// applies to stdout/stderr output.
    #[arg(long, env = "VECTOR_INTERNAL_LOGS_SOURCE_RATE_LIMIT")]
    pub internal_logs_source_rate_limit: Option<NonZeroU64>,

    /// Set the duration in seconds to wait for graceful shutdown after SIGINT or SIGTERM are
    /// received. After the duration has passed, Vector will force shutdown. To never force
    /// shutdown, use `--no-graceful-shutdown-limit`.
    #[arg(
        long,
        default_value = "60",
        env = "VECTOR_GRACEFUL_SHUTDOWN_LIMIT_SECS",
        group = "graceful-shutdown-limit"
    )]
    pub graceful_shutdown_limit_secs: NonZeroU64,

    /// Never time out while waiting for graceful shutdown after SIGINT or SIGTERM received.
    /// This is useful when you would like for Vector to attempt to send data until terminated
    /// by a SIGKILL. Overrides/cannot be set with `--graceful-shutdown-limit-secs`.
    #[arg(
        long,
        default_value = "false",
        env = "VECTOR_NO_GRACEFUL_SHUTDOWN_LIMIT",
        group = "graceful-shutdown-limit"
    )]
    pub no_graceful_shutdown_limit: bool,

    /// Set runtime allocation tracing
    #[cfg(all(unix, feature = "tikv-jemallocator"))]
    #[arg(long, env = "ALLOCATION_TRACING", default_value = "false")]
    pub allocation_tracing: bool,

    /// Set allocation tracing reporting rate in milliseconds.
    #[cfg(all(unix, feature = "tikv-jemallocator"))]
    #[arg(
        long,
        env = "ALLOCATION_TRACING_REPORTING_INTERVAL_MS",
        default_value = "5000"
    )]
    pub allocation_tracing_reporting_interval_ms: u64,

    /// Disable probing and configuration of root certificate locations on the system for OpenSSL.
    ///
    /// The probe functionality manipulates the `SSL_CERT_FILE` and `SSL_CERT_DIR` environment variables
    /// in the Vector process. This behavior can be problematic for users of the `exec` source, which by
    /// default inherits the environment of the Vector process.
    #[arg(long, env = "VECTOR_OPENSSL_NO_PROBE", default_value = "false")]
    pub openssl_no_probe: bool,

    /// Allow the configuration to run without any components. This is useful for loading in an
    /// empty stub config that will later be replaced with actual components. Note that this is
    /// likely not useful without also watching for config file changes as described in
    /// `--watch-config`.
    #[arg(long, env = "VECTOR_ALLOW_EMPTY_CONFIG", default_value = "false")]
    pub allow_empty_config: bool,

    /// Maximum number of bytes allowed after decompressing a payload.
    ///
    /// Sources that decompress incoming payloads (gzip, deflate, zstd, snappy) enforce this cap to
    /// prevent a compressed "bomb" from exhausting memory. Payloads whose decompressed size exceeds
    /// the limit are rejected.
    ///
    /// Defaults to 104857600 (100 MiB). Raise this only when sources routinely receive
    /// legitimately large compressed payloads.
    #[arg(
        long,
        env = "VECTOR_MAX_DECOMPRESSED_SIZE_BYTES",
        default_value = "104857600"
    )]
    pub max_decompressed_size_bytes: usize,

    /// Raise the file descriptor soft limit (RLIMIT_NOFILE) to the hard limit at startup.
    ///
    /// Many systems default the soft limit to 1024 (Linux) or 256 (macOS), which is too low
    /// when Vector monitors large numbers of log files. This flag raises the soft limit to
    /// prevent "Too many open files" errors without requiring manual sysadmin intervention.
    #[cfg(unix)]
    #[arg(long, env = "VECTOR_RAISE_FD_LIMIT", default_value = "false")]
    pub raise_fd_limit: bool,
}

impl RootOpts {
    /// Return a list of config paths with the associated formats.
    pub fn config_paths_with_formats(&self) -> Vec<config::ConfigPath> {
        config::merge_path_lists(vec![
            (&self.config_paths, None),
            (&self.config_paths_toml, Some(config::Format::Toml)),
            (&self.config_paths_json, Some(config::Format::Json)),
            (&self.config_paths_yaml, Some(config::Format::Yaml)),
        ])
        .map(|(path, hint)| config::ConfigPath::File(path, hint))
        .chain(
            self.config_dirs
                .iter()
                .map(|dir| config::ConfigPath::Dir(dir.to_path_buf())),
        )
        .collect()
    }

    pub fn init_global(&self) {
        if !self.openssl_no_probe {
            unsafe {
                openssl_probe::init_openssl_env_vars();
            }
        }

        crate::metrics::init_global().expect("metrics initialization failed");
    }
}

/// Raise the soft file descriptor limit (RLIMIT_NOFILE) as high as the OS allows.
///
/// Many systems default the soft limit to 1024 (Linux) or 256 (macOS), which is too low
/// for Vector when it monitors large numbers of log files. Raising it prevents
/// "Too many open files (os error 24)" errors without requiring manual sysadmin intervention.
///
/// On Linux, the soft limit is raised to the hard limit (typically 65536+).
/// On macOS, the hard limit can be RLIM_INFINITY, so we first try the hard limit,
/// then fall back to the kernel-enforced `kern.maxfilesperproc` (typically 10240).
#[cfg(unix)]
pub(crate) fn raise_file_descriptor_limit() {
    use nix::sys::resource::{Resource, getrlimit, setrlimit};
    use tracing::{info, warn};

    let (soft, hard) = match getrlimit(Resource::RLIMIT_NOFILE) {
        Ok(limits) => limits,
        Err(err) => {
            warn!(message = "Failed to get file descriptor limit.", %err);
            return;
        }
    };

    if soft >= hard {
        return; // Already at maximum
    }

    // Try setting soft limit to hard limit (works on Linux, may fail on macOS)
    if setrlimit(Resource::RLIMIT_NOFILE, hard, hard).is_ok() {
        info!(
            message = "Raised file descriptor limit.",
            from = soft,
            to = hard,
        );
        return;
    }

    // On macOS, the hard limit can be RLIM_INFINITY which setrlimit rejects.
    // Fall back to the kernel-enforced kern.maxfilesperproc.
    #[cfg(target_os = "macos")]
    {
        if let Some(maxfiles) = macos_maxfilesperproc()
            && maxfiles > soft
            && setrlimit(Resource::RLIMIT_NOFILE, maxfiles, hard).is_ok()
        {
            info!(
                message = "Raised file descriptor limit.",
                from = soft,
                to = maxfiles,
            );
            return;
        }
    }

    warn!(
        message = "Failed to raise file descriptor limit.",
        current = soft,
        attempted = hard,
    );
}

/// Query the macOS kernel limit on per-process open files.
#[cfg(target_os = "macos")]
fn macos_maxfilesperproc() -> Option<libc::rlim_t> {
    let mut maxfiles: libc::c_int = 0;
    let mut len = std::mem::size_of::<libc::c_int>() as libc::size_t;
    // Safety: sysctlbyname with a valid null-terminated name and correctly sized output buffer.
    // No safe wrapper exists for this macOS-specific call.
    let ret = unsafe {
        libc::sysctlbyname(
            c"kern.maxfilesperproc".as_ptr(),
            &mut maxfiles as *mut libc::c_int as *mut libc::c_void,
            &mut len,
            std::ptr::null_mut(),
            0,
        )
    };
    if ret == 0 && maxfiles > 0 {
        Some(maxfiles as libc::rlim_t)
    } else {
        None
    }
}

#[derive(Parser, Debug)]
#[command(rename_all = "kebab-case")]
pub enum SubCommand {
    /// Validate the target config, then exit.
    Validate(validate::Opts),

    /// Convert a config file from one format to another.
    /// This command can also walk directories recursively and convert all config files that are discovered.
    /// Note that this is a best effort conversion due to the following reasons:
    /// * The comments from the original config file are not preserved.
    /// * Explicitly set default values in the original implementation might be omitted.
    /// * Depending on how each source/sink config struct configures serde, there might be entries with null values.
    ConvertConfig(convert_config::Opts),

    /// Generate a Vector configuration containing a list of components.
    Generate(generate::Opts),

    /// Generate the configuration schema for this version of Vector. (experimental)
    ///
    /// A JSON Schema document will be generated that represents the valid schema for a
    /// Vector configuration. This schema is based on the "full" configuration, such that for usages
    /// where a configuration is split into multiple files, the schema would apply to those files
    /// only when concatenated together.
    ///
    /// By default all output is written to stdout. The `output_path` option can be used to redirect to a file.
    GenerateSchema(generate_schema::Opts),

    /// Generate shell completion, then exit.
    #[command(hide = true)]
    Completion(completion::Opts),

    /// Output a provided Vector configuration file/dir as a single JSON object, useful for checking in to version control.
    #[command(hide = true)]
    Config(config::Opts),

    /// List available components, then exit.
    List(list::Opts),

    /// Run Vector config unit tests, then exit. This command is experimental and therefore subject to change.
    /// For guidance on how to write unit tests check out <https://vector.dev/guides/level-up/unit-testing/>.
    Test(unit_test::Opts),

    /// Output the topology as visual representation using the DOT language which can be rendered by GraphViz
    Graph(graph::Opts),

    /// Display topology and metrics in the console, for a local or remote Vector instance
    #[cfg(feature = "top")]
    Top(top::Opts),

    /// Observe output log events from source or transform components. Logs are sampled at a specified interval.
    #[cfg(feature = "api-client")]
    Tap(tap::Opts),

    /// Manage the vector service.
    #[cfg(windows)]
    Service(service::Opts),

    /// Vector Remap Language CLI
    Vrl(vrl::cli::Opts),
}

impl SubCommand {
    #[expect(
        clippy::missing_const_for_fn,
        reason = "the #[cfg(windows)] arm calls a non-const method"
    )]
    pub fn dangerously_allow_env_var_interpolation(&self) -> bool {
        match self {
            Self::Config(c) => c.dangerously_allow_env_var_interpolation,
            Self::Graph(g) => g.dangerously_allow_env_var_interpolation,
            Self::Test(t) => t.dangerously_allow_env_var_interpolation,
            Self::Validate(v) => v.dangerously_allow_env_var_interpolation,
            #[cfg(windows)]
            Self::Service(s) => s.dangerously_allow_env_var_interpolation(),
            _ => false,
        }
    }

    pub async fn execute(
        &self,
        mut signals: signal::SignalPair,
        color: bool,
    ) -> exitcode::ExitCode {
        match self {
            Self::Completion(s) => completion::cmd(s),
            Self::Config(c) => config::cmd(c),
            Self::ConvertConfig(opts) => convert_config::cmd(opts),
            Self::Generate(g) => generate::cmd(g),
            Self::GenerateSchema(opts) => generate_schema::cmd(opts),
            Self::Graph(g) => graph::cmd(g),
            Self::List(l) => list::cmd(l),
            #[cfg(windows)]
            Self::Service(s) => service::cmd(s),
            #[cfg(feature = "api-client")]
            Self::Tap(t) => tap::cmd(t, signals.receiver).await,
            Self::Test(t) => unit_test::cmd(t, &mut signals.handler).await,
            #[cfg(feature = "top")]
            Self::Top(t) => top::cmd(t).await,
            Self::Validate(v) => validate::validate(v, color).await,
            Self::Vrl(s) => vrl::cli::cmd::cmd(s, vector_vrl_functions::all()),
        }
    }
}

#[derive(clap::ValueEnum, Debug, Clone, Copy, PartialEq, Eq)]
pub enum Color {
    Auto,
    Always,
    Never,
}

impl Color {
    pub fn use_color(&self) -> bool {
        match self {
            #[cfg(unix)]
            Color::Auto => {
                use std::io::IsTerminal;
                std::io::stdout().is_terminal()
            }
            #[cfg(windows)]
            Color::Auto => false, // ANSI colors are not supported by cmd.exe
            Color::Always => true,
            Color::Never => false,
        }
    }
}

#[derive(clap::ValueEnum, Debug, Clone, Copy, PartialEq, Eq)]
pub enum LogFormat {
    Text,
    Json,
}

#[derive(clap::ValueEnum, Debug, Clone, Copy, PartialEq, Eq)]
pub enum WatchConfigMethod {
    /// Recommended watcher for the current OS, usually `inotify` for Linux-based systems.
    Recommended,
    /// Poll-based watcher, typically used for watching files on EFS/NFS-like network storage systems.
    /// The interval is determined by  [`RootOpts::watch_config_poll_interval_seconds`].
    Poll,
}

pub fn handle_config_errors(errors: Vec<String>) -> exitcode::ExitCode {
    for error in errors {
        error!(message = "Configuration error.", %error, internal_log_rate_limit = false);
    }

    exitcode::CONFIG
}

#[cfg(test)]
mod tests {
    #[cfg(unix)]
    fn run_in_subprocess(test_name: &str) {
        let exe = std::env::current_exe().unwrap();
        let output = std::process::Command::new(exe)
            .env("__VECTOR_SUBPROCESS_TEST", "1")
            .args(["--exact", test_name, "--nocapture"])
            .output()
            .unwrap();
        assert!(
            output.status.success(),
            "subprocess test failed:\nstdout: {}\nstderr: {}",
            String::from_utf8_lossy(&output.stdout),
            String::from_utf8_lossy(&output.stderr),
        );
    }

    #[test]
    #[cfg(unix)]
    fn test_raise_file_descriptor_limit() {
        if std::env::var("__VECTOR_SUBPROCESS_TEST").is_err() {
            run_in_subprocess("cli::tests::test_raise_file_descriptor_limit");
            return;
        }

        use nix::sys::resource::{Resource, getrlimit, setrlimit};

        let (original_soft, hard) = getrlimit(Resource::RLIMIT_NOFILE).unwrap();
        let lowered = std::cmp::min(original_soft, 256);
        if lowered < hard {
            setrlimit(Resource::RLIMIT_NOFILE, lowered, hard).unwrap();

            let (soft_before, _) = getrlimit(Resource::RLIMIT_NOFILE).unwrap();
            assert_eq!(soft_before, lowered);

            super::raise_file_descriptor_limit();

            let (soft_after, _) = getrlimit(Resource::RLIMIT_NOFILE).unwrap();
            assert!(
                soft_after > lowered,
                "Expected soft limit to be raised above {lowered}, got {soft_after}"
            );
        }
    }

    #[test]
    #[cfg(unix)]
    fn test_raise_file_descriptor_limit_already_at_max() {
        if std::env::var("__VECTOR_SUBPROCESS_TEST").is_err() {
            run_in_subprocess("cli::tests::test_raise_file_descriptor_limit_already_at_max");
            return;
        }

        use nix::sys::resource::{Resource, getrlimit, setrlimit};

        let (_, hard) = getrlimit(Resource::RLIMIT_NOFILE).unwrap();

        if setrlimit(Resource::RLIMIT_NOFILE, hard, hard).is_err() {
            #[cfg(target_os = "macos")]
            if let Some(maxfiles) = super::macos_maxfilesperproc() {
                setrlimit(Resource::RLIMIT_NOFILE, maxfiles, hard).ok();
            }
        }

        let (soft_before, _) = getrlimit(Resource::RLIMIT_NOFILE).unwrap();

        super::raise_file_descriptor_limit();

        let (soft_after, _) = getrlimit(Resource::RLIMIT_NOFILE).unwrap();
        assert_eq!(soft_before, soft_after);
    }

    #[test]
    #[cfg(target_os = "macos")]
    fn test_macos_maxfilesperproc_returns_positive() {
        let result = super::macos_maxfilesperproc();
        assert!(
            result.is_some(),
            "macos_maxfilesperproc() should return Some on macOS"
        );
        assert!(
            result.unwrap() > 0,
            "kern.maxfilesperproc should be positive"
        );
    }
}