[reconfigurator] Retrieve keeper lgif information

clickhouse-admin/Cargo.toml

@@ -30,9 +30,13 @@ toml.workspace = true
 omicron-workspace-hack.workspace = true
 
 [dev-dependencies]
+clickward.workspace = true
+dropshot.workspace = true
 expectorate.workspace = true
 nexus-test-utils.workspace = true
 omicron-test-utils.workspace = true
+oximeter-db.workspace = true
+oximeter-test-utils.workspace = true
 openapi-lint.workspace = true
 openapiv3.workspace = true
 serde_json.workspace = true

clickhouse-admin/api/src/lib.rs

@@ -3,8 +3,10 @@
 // file, You can obtain one at https://mozilla.org/MPL/2.0/.
 
 use clickhouse_admin_types::config::{KeeperConfig, ReplicaConfig};
-use clickhouse_admin_types::{KeeperSettings, ServerSettings};
-use dropshot::{HttpError, HttpResponseCreated, RequestContext, TypedBody};
+use clickhouse_admin_types::{KeeperSettings, Lgif, ServerSettings};
+use dropshot::{
+    HttpError, HttpResponseCreated, HttpResponseOk, RequestContext, TypedBody,
+};
 use omicron_common::api::external::Generation;
 use schemars::JsonSchema;
 use serde::Deserialize;
@@ -50,4 +52,15 @@ pub trait ClickhouseAdminApi {
         rqctx: RequestContext<Self::Context>,
         body: TypedBody<KeeperConfigurableSettings>,
     ) -> Result<HttpResponseCreated<KeeperConfig>, HttpError>;
+
+    /// Retrieve a logically grouped information file from a keeper node.
+    /// This information is used internally by ZooKeeper to manage snapshots
+    /// and logs for consistency and recovery.
+    #[endpoint {
+        method = GET,
+        path = "/keeper/lgif",
+    }]
+    async fn lgif(
+        rqctx: RequestContext<Self::Context>,
+    ) -> Result<HttpResponseOk<Lgif>, HttpError>;
 }

clickhouse-admin/src/bin/clickhouse-admin.rs

@@ -7,7 +7,7 @@
 use anyhow::anyhow;
 use camino::Utf8PathBuf;
 use clap::Parser;
-use omicron_clickhouse_admin::{Clickward, Config};
+use omicron_clickhouse_admin::{ClickhouseCli, Clickward, Config};
 use omicron_common::cmd::fatal;
 use omicron_common::cmd::CmdError;
 use std::net::{SocketAddr, SocketAddrV6};
@@ -27,6 +27,14 @@ enum Args {
         /// Path to the server configuration file
         #[clap(long, short, action)]
         config: Utf8PathBuf,
+
+        /// Address on which the clickhouse server or keeper is listening on
+        #[clap(long, short = 'l', action)]
+        listen_address: SocketAddrV6,
+
+        /// Path to the clickhouse binary
+        #[clap(long, short, action)]
+        binary_path: Utf8PathBuf,
     },
 }
 
@@ -41,17 +49,21 @@ async fn main_impl() -> Result<(), CmdError> {
     let args = Args::parse();
 
     match args {
-        Args::Run { http_address, config } => {
+        Args::Run { http_address, config, listen_address, binary_path } => {
             let mut config = Config::from_file(&config)
                 .map_err(|err| CmdError::Failure(anyhow!(err)))?;
             config.dropshot.bind_address = SocketAddr::V6(http_address);
-
             let clickward = Clickward::new();
+            let clickhouse_cli =
+                ClickhouseCli::new(binary_path, listen_address);
 
-            let server =
-                omicron_clickhouse_admin::start_server(clickward, config)
-                    .await
-                    .map_err(|err| CmdError::Failure(anyhow!(err)))?;
+            let server = omicron_clickhouse_admin::start_server(
+                clickward,
+                clickhouse_cli,
+                config,
+            )
+            .await
+            .map_err(|err| CmdError::Failure(anyhow!(err)))?;
             server.await.map_err(|err| {
                 CmdError::Failure(anyhow!(
                     "server failed after starting: {err}"

clickhouse-admin/src/clickhouse_cli.rs

@@ -0,0 +1,130 @@
+// This Source Code Form is subject to the terms of the Mozilla Public
+// License, v. 2.0. If a copy of the MPL was not distributed with this
+// file, You can obtain one at https://mozilla.org/MPL/2.0/.
+
+use anyhow::Result;
+use camino::Utf8PathBuf;
+use clickhouse_admin_types::Lgif;
+use dropshot::HttpError;
+use illumos_utils::{output_to_exec_error, ExecutionError};
+use slog::Logger;
+use slog_error_chain::{InlineErrorChain, SlogInlineError};
+use std::ffi::OsStr;
+use std::io;
+use std::net::SocketAddrV6;
+use tokio::process::Command;
+
+#[derive(Debug, thiserror::Error, SlogInlineError)]
+pub enum ClickhouseCliError {
+    #[error("failed to run `clickhouse {subcommand}`")]
+    Run {
+        description: &'static str,
+        subcommand: String,
+        #[source]
+        err: io::Error,
+    },
+    #[error(transparent)]
+    ExecutionError(#[from] ExecutionError),
+    #[error("failed to parse command output")]
+    Parse {
+        description: &'static str,
+        stdout: String,
+        stderr: String,
+        #[source]
+        err: anyhow::Error,
+    },
+}
+
+impl From<ClickhouseCliError> for HttpError {
+    fn from(err: ClickhouseCliError) -> Self {
+        match err {
+            ClickhouseCliError::Run { .. }
+            | ClickhouseCliError::Parse { .. }
+            | ClickhouseCliError::ExecutionError(_) => {
+                let message = InlineErrorChain::new(&err).to_string();
+                HttpError {
+                    status_code: http::StatusCode::INTERNAL_SERVER_ERROR,
+                    error_code: Some(String::from("Internal")),
+                    external_message: message.clone(),
+                    internal_message: message,
+                }
+            }
+        }
+    }
+}
+
+#[derive(Debug)]
+pub struct ClickhouseCli {
+    /// Path to where the clickhouse binary is located
+    pub binary_path: Utf8PathBuf,
+    /// Address on where the clickhouse keeper is listening on
+    pub listen_address: SocketAddrV6,
+    pub log: Option<Logger>,
+}
+
+impl ClickhouseCli {
+    pub fn new(binary_path: Utf8PathBuf, listen_address: SocketAddrV6) -> Self {
+        Self { binary_path, listen_address, log: None }
+    }
+
+    pub fn with_log(mut self, log: Logger) -> Self {
+        self.log = Some(log);
+        self
+    }
+
+    pub async fn lgif(&self) -> Result<Lgif, ClickhouseCliError> {
+        self.keeper_client_non_interactive(
+            "lgif",
+            "Retrieve logically grouped information file",
+            Lgif::parse,
+            self.log.clone().unwrap(),
+        )
+        .await
+    }
+
+    async fn keeper_client_non_interactive<F, T>(
+        &self,
+        query: &str,
+        subcommand_description: &'static str,
+        parse: F,
+        log: Logger,
+    ) -> Result<T, ClickhouseCliError>
+    where
+        F: FnOnce(&Logger, &[u8]) -> Result<T>,
+    {
+        let mut command = Command::new(&self.binary_path);
+        command
+            .arg("keeper-client")
+            .arg("--host")
+            .arg(&format!("[{}]", self.listen_address.ip()))
+            .arg("--port")
+            .arg(&format!("{}", self.listen_address.port()))
+            .arg("--query")
+            .arg(query);
+
+        let output = command.output().await.map_err(|err| {
+            let err_args: Vec<&OsStr> = command.as_std().get_args().collect();
+            let err_args_parsed: Vec<String> = err_args
+                .iter()
+                .map(|&os_str| os_str.to_string_lossy().into_owned())
+                .collect();
+            let err_args_str = err_args_parsed.join(" ");
+            ClickhouseCliError::Run {
+                description: subcommand_description,
+                subcommand: err_args_str,
+                err,
+            }
+        })?;
+
+        if !output.status.success() {
+            return Err(output_to_exec_error(command.as_std(), &output).into());
+        }
+
+        parse(&log, &output.stdout).map_err(|err| ClickhouseCliError::Parse {
+            description: subcommand_description,
+            stdout: String::from_utf8_lossy(&output.stdout).to_string(),
+            stderr: String::from_utf8_lossy(&output.stdout).to_string(),
+            err,
+        })
+    }
+}

clickhouse-admin/src/context.rs

@@ -2,20 +2,29 @@
 // License, v. 2.0. If a copy of the MPL was not distributed with this
 // file, You can obtain one at https://mozilla.org/MPL/2.0/.
 
-use crate::Clickward;
+use crate::{ClickhouseCli, Clickward};
 use slog::Logger;
 
 pub struct ServerContext {
     clickward: Clickward,
+    clickhouse_cli: ClickhouseCli,
     _log: Logger,
 }
 
 impl ServerContext {
-    pub fn new(clickward: Clickward, _log: Logger) -> Self {
-        Self { clickward, _log }
+    pub fn new(
+        clickward: Clickward,
+        clickhouse_cli: ClickhouseCli,
+        _log: Logger,
+    ) -> Self {
+        Self { clickward, clickhouse_cli, _log }
     }
 
     pub fn clickward(&self) -> &Clickward {
         &self.clickward
     }
+
+    pub fn clickhouse_cli(&self) -> &ClickhouseCli {
+        &self.clickhouse_cli
+    }
 }

clickhouse-admin/src/http_entrypoints.rs

@@ -5,7 +5,10 @@
 use crate::context::ServerContext;
 use clickhouse_admin_api::*;
 use clickhouse_admin_types::config::{KeeperConfig, ReplicaConfig};
-use dropshot::{HttpError, HttpResponseCreated, RequestContext, TypedBody};
+use clickhouse_admin_types::Lgif;
+use dropshot::{
+    HttpError, HttpResponseCreated, HttpResponseOk, RequestContext, TypedBody,
+};
 use std::sync::Arc;
 
 type ClickhouseApiDescription = dropshot::ApiDescription<Arc<ServerContext>>;
@@ -44,4 +47,12 @@ impl ClickhouseAdminApi for ClickhouseAdminImpl {
         let output = ctx.clickward().generate_keeper_config(keeper.settings)?;
         Ok(HttpResponseCreated(output))
     }
+
+    async fn lgif(
+        rqctx: RequestContext<Self::Context>,
+    ) -> Result<HttpResponseOk<Lgif>, HttpError> {
+        let ctx = rqctx.context();
+        let output = ctx.clickhouse_cli().lgif().await?;
+        Ok(HttpResponseOk(output))
+    }
 }

clickhouse-admin/src/lib.rs

@@ -11,11 +11,13 @@ use std::error::Error;
 use std::io;
 use std::sync::Arc;
 
+mod clickhouse_cli;
 mod clickward;
 mod config;
 mod context;
 mod http_entrypoints;
 
+pub use clickhouse_cli::ClickhouseCli;
 pub use clickward::Clickward;
 pub use config::Config;
 
@@ -34,6 +36,7 @@ pub type Server = dropshot::HttpServer<Arc<ServerContext>>;
 /// Start the dropshot server
 pub async fn start_server(
     clickward: Clickward,
+    clickhouse_cli: ClickhouseCli,
     server_config: Config,
 ) -> Result<Server, StartError> {
     let (drain, registration) = slog_dtrace::with_drain(
@@ -56,6 +59,8 @@ pub async fn start_server(
 
     let context = ServerContext::new(
         clickward,
+        clickhouse_cli
+            .with_log(log.new(slog::o!("component" => "ClickhouseCli"))),
         log.new(slog::o!("component" => "ServerContext")),
     );
     let http_server_starter = dropshot::HttpServerStarter::new(