pluto: use apiclient API rather than CLI

Replace the calls to shell out to `apiclient` with direct API calls.
Add logging so that we can better monitor time spent on each portion of pluto.

Author: fletcherw

sources/Cargo.lock

@@ -13,6 +13,7 @@ exclude = ["README.md"]
 fips = ["aws-lc-rs/fips", "aws-smithy-experimental/crypto-aws-lc-fips", "rustls/fips"]
 
 [dependencies]
+apiclient.workspace = true
 aws-config.workspace = true
 aws-lc-rs = { workspace = true, features = ["bindgen"] }
 aws-sdk-eks.workspace = true
@@ -24,10 +25,13 @@ base64.workspace = true
 bottlerocket-modeled-types.workspace = true
 bottlerocket-settings-models.workspace = true
 constants.workspace = true
+http.workspace = true
 imdsclient.workspace = true
+log.workspace = true
 rustls.workspace = true
 serde = { workspace = true, features = ["derive"] }
 serde_json.workspace = true
+simplelog.workspace = true
 snafu.workspace = true
 tempfile.workspace = true
 tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }

sources/api/pluto/Cargo.toml

@@ -1,8 +1,6 @@
 use bottlerocket_settings_models::{AwsSettingsV1, KubernetesSettingsV1, NetworkSettingsV1};
 use serde::Deserialize;
-use snafu::{ensure, ResultExt, Snafu};
-use std::ffi::OsStr;
-use tokio::process::Command;
+use snafu::{ResultExt, Snafu};
 
 /// The result type for the [`api`] module.
 pub(super) type Result<T> = std::result::Result<T, Error>;
@@ -128,53 +126,56 @@ struct APISettingsResponse {
 
 #[derive(Debug, Snafu)]
 pub(crate) enum Error {
-    #[snafu(display("Failed to call apiclient: {}", source))]
-    CommandFailure { source: std::io::Error },
-    #[snafu(display("apiclient execution failed: {}", reason))]
-    ExecutionFailure { reason: String },
+    #[snafu(display("Failed to call API get_prefixes: {}", source))]
+    GetPrefix { source: apiclient::get::Error },
     #[snafu(display("Deserialization of configuration file failed: {}", source))]
     Deserialize {
         #[snafu(source(from(serde_json::Error, Box::new)))]
         source: Box<serde_json::Error>,
     },
-}
-
-pub(crate) async fn client_command<I, S>(args: I) -> Result<Vec<u8>>
-where
-    I: IntoIterator<Item = S>,
-    S: AsRef<OsStr>,
-{
-    let result = Command::new("/usr/bin/apiclient")
-        .args(args)
-        .output()
-        .await
-        .context(CommandFailureSnafu)?;
-
-    ensure!(
-        result.status.success(),
-        ExecutionFailureSnafu {
-            reason: String::from_utf8_lossy(&result.stderr)
-        }
-    );
+    #[snafu(display("Setting settings values in API failed: {}", source))]
+    SetSettings { source: apiclient::Error },
 
-    Ok(result.stdout)
+    #[snafu(display("Failed to serialize generated settings: {}", source))]
+    Serialize { source: serde_json::Error },
 }
 
 /// Gets the info that we need to know about the EKS cluster from the Bottlerocket API.
 pub(crate) async fn get_aws_k8s_info() -> Result<SettingsView> {
-    let view_str = client_command(&[
-        "get",
-        "settings.aws",
-        "settings.network",
-        "settings.kubernetes",
-    ])
-    .await?;
+    let prefixes = ["settings.aws", "settings.network", "settings.kubernetes"]
+        .into_iter()
+        .map(str::to_string)
+        .collect();
+    let response = apiclient::get::get_prefixes(constants::API_SOCKET, prefixes)
+        .await
+        .context(GetPrefixSnafu)?;
 
     let api_response: APISettingsResponse =
-        serde_json::from_slice(view_str.as_slice()).context(DeserializeSnafu)?;
+        serde_json::from_value::<APISettingsResponse>(response).context(DeserializeSnafu)?;
     Ok(api_response.settings)
 }
 
+/// Send the settings to the datastore through the API
+pub(crate) async fn set_settings(settings: &KubernetesSettingsV1) -> Result<()> {
+    let generated_settings = serde_json::json!({
+        "kubernetes": serde_json::to_value(settings).context(SerializeSnafu)?
+    });
+    let request_body = generated_settings.to_string();
+
+    let uri = &format!(
+        "{}?tx={}",
+        constants::API_SETTINGS_URI,
+        constants::LAUNCH_TRANSACTION,
+    );
+    let method = "PATCH";
+    // Ignore response code/body, raw_request already checks for success.
+    let _ = apiclient::raw_request(constants::API_SOCKET, uri, method, Some(request_body))
+        .await
+        .context(SetSettingsSnafu)?;
+
+    Ok(())
+}
+
 #[cfg(test)]
 mod test {
     use super::*;

sources/api/pluto/src/api.rs

@@ -36,11 +36,15 @@ mod aws;
 mod ec2;
 mod eks;
 
+#[macro_use]
+extern crate log;
+
 use api::{settings_view_get, settings_view_set, SettingsViewDelta};
 use aws_smithy_experimental::hyper_1_0::CryptoMode;
 use base64::Engine;
 use bottlerocket_modeled_types::{KubernetesClusterDnsIp, KubernetesHostnameOverrideSource};
 use imdsclient::ImdsClient;
+use simplelog::{Config as LogConfig, LevelFilter, SimpleLogger};
 use snafu::{ensure, OptionExt, ResultExt};
 use std::fs::File;
 use std::io::{BufRead, BufReader, Write};
@@ -136,9 +140,6 @@ mod error {
         #[snafu(display("Failed to read line: {}", source))]
         IoReadLine { source: std::io::Error },
 
-        #[snafu(display("Failed to serialize generated settings: {}", source))]
-        Serialize { source: serde_json::Error },
-
         #[snafu(display("Failed to set generated settings: {}", source))]
         SetFailure { source: api::Error },
 
@@ -162,6 +163,9 @@ mod error {
 
         #[snafu(display("Unable to create tempdir: {}", source))]
         Tempdir { source: std::io::Error },
+
+        #[snafu(display("Logger setup error: {}", source))]
+        Logger { source: log::SetLoggerError },
     }
 }
 
@@ -488,6 +492,11 @@ fn set_aws_config(aws_k8s_info: &SettingsViewDelta, filepath: &Path) -> Result<(
 }
 
 async fn run() -> Result<()> {
+    // SimpleLogger will send errors to stderr and anything less to stdout.
+    SimpleLogger::init(LevelFilter::Info, LogConfig::default()).context(error::LoggerSnafu)?;
+    info!("Pluto started");
+
+    info!("Retrieving settings values");
     let mut client = ImdsClient::new();
     let current_settings = api::get_aws_k8s_info().await.context(error::AwsInfoSnafu)?;
     let mut aws_k8s_info = SettingsViewDelta::from_api_response(current_settings);
@@ -498,23 +507,16 @@ async fn run() -> Result<()> {
     let aws_config_file_path = temp_dir.path().join(AWS_CONFIG_FILE);
     set_aws_config(&aws_k8s_info, Path::new(&aws_config_file_path))?;
 
+    info!("Generating kubernetes settings");
     generate_cluster_dns_ip(&mut client, &mut aws_k8s_info).await?;
     generate_node_ip(&mut client, &mut aws_k8s_info).await?;
     generate_max_pods(&mut client, &mut aws_k8s_info).await?;
     generate_provider_id(&mut client, &mut aws_k8s_info).await?;
     generate_node_name(&mut client, &mut aws_k8s_info).await?;
 
     if let Some(k8s_settings) = &aws_k8s_info.delta().kubernetes {
-        let generated_settings = serde_json::json!({
-            "kubernetes": serde_json::to_value(k8s_settings).context(error::SerializeSnafu)?
-        });
-        let json_str = generated_settings.to_string();
-        let uri = &format!(
-            "{}?tx={}",
-            constants::API_SETTINGS_URI,
-            constants::LAUNCH_TRANSACTION
-        );
-        api::client_command(&["raw", "-m", "PATCH", "-u", uri, "-d", json_str.as_str()])
+        info!("Sending settings values to the API");
+        api::set_settings(k8s_settings)
             .await
             .context(error::SetFailureSnafu)?;
     }