mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-21 01:53:30 +00:00
Add a local environment inventory command (#7728)
feat(connect): add local environment inventory command
This commit is contained in:
Chris
@@ -1,3 +1,4 @@
|
||||
58d61bb10b9443eca3c0e9dc340cbe91144eed32a705617cc3406bd825439824 accept-vectors.json
|
||||
d1a73b0a348845bf3ed9fb68301babc5abbc6e72243a610db1cd4794b1328070 canonical-hash.json
|
||||
f1107d3e6accbaee468f1a1fdb79d7103fb2aadafd85c39020fbbca5173b03b4 field-registry.json
|
||||
b3b2e7f761198d4823c94440637a48153437183f4cacec5118570e1920f73b29 old-agent-vectors.json
|
||||
|
||||
@@ -0,0 +1,58 @@
|
||||
{
|
||||
"protocolVersion": "v1",
|
||||
"fixtureSet": "inventory",
|
||||
"fixture": "accept-vectors",
|
||||
"description": "Producer vectors for the negotiated inventory.environment@1 summary. The input names source observations used by the RustFS collector; output is the complete identifier-free payload allowed to leave the deployment.",
|
||||
"schemaVersion": 1,
|
||||
"capability": "inventory.environment@1",
|
||||
"vectors": [
|
||||
{
|
||||
"name": "known two-node deployment",
|
||||
"input": {
|
||||
"persistedInventory": { "nodeCount": 2, "driveCount": 8 },
|
||||
"sourceObservation": { "osFamily": "LINUX", "filesystemTypes": ["xfs"] }
|
||||
},
|
||||
"expected": {
|
||||
"decision": "ACCEPT",
|
||||
"output": { "nodeCount": 2, "driveCount": 8, "osFamily": "LINUX", "filesystemTypes": ["xfs"] }
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "secret-like filesystem observations are not exported",
|
||||
"input": {
|
||||
"persistedInventory": { "nodeCount": 2, "driveCount": 8 },
|
||||
"sourceObservation": {
|
||||
"osFamily": "LINUX",
|
||||
"filesystemTypes": ["xfs"],
|
||||
"mountPath": "/srv/synthetic-customer-a",
|
||||
"mountOptions": "rw,password=SYNTHETIC_NOT_A_REAL_SECRET",
|
||||
"deviceLabel": "synthetic-customer-volume",
|
||||
"credential": "SYNTHETIC_NOT_A_REAL_CREDENTIAL"
|
||||
}
|
||||
},
|
||||
"expected": {
|
||||
"decision": "ACCEPT",
|
||||
"discarded": ["credential", "deviceLabel", "mountOptions", "mountPath"],
|
||||
"output": { "nodeCount": 2, "driveCount": 8, "osFamily": "LINUX", "filesystemTypes": ["xfs"] }
|
||||
}
|
||||
},
|
||||
{
|
||||
"name": "old schema is rejected before collection",
|
||||
"input": { "schemaVersion": 0, "capability": "inventory.environment@1" },
|
||||
"expected": { "decision": "REJECT", "producerError": "inventory_environment_unsupported_version" }
|
||||
},
|
||||
{
|
||||
"name": "unknown capability is rejected before collection",
|
||||
"input": { "schemaVersion": 1, "capability": "inventory.environment@2" },
|
||||
"expected": { "decision": "REJECT", "producerError": "inventory_environment_unsupported_capability" }
|
||||
},
|
||||
{
|
||||
"name": "missing filesystem source is not replaced with an empty success",
|
||||
"input": {
|
||||
"persistedInventory": { "nodeCount": 2, "driveCount": 8 },
|
||||
"sourceObservation": { "osFamily": "LINUX", "filesystemTypes": [] }
|
||||
},
|
||||
"expected": { "decision": "UNSUPPORTED", "producerError": "inventory_environment_source_unavailable", "output": null }
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -127,6 +127,8 @@ pub enum ConnectCommands {
|
||||
Register(ConnectRegisterOpts),
|
||||
/// Import, verify, or inspect a signed Connect service license
|
||||
License(ConnectLicenseOpts),
|
||||
/// Read the persisted deployment inventory and collect an approved environment summary
|
||||
Inventory(ConnectInventoryOpts),
|
||||
/// Run an explicitly approved, bounded local performance measurement
|
||||
Performance(ConnectPerformanceOpts),
|
||||
/// Capture a consent-bound local profile and write a signed export
|
||||
@@ -139,6 +141,37 @@ pub enum ConnectCommands {
|
||||
Top(ConnectTopOpts),
|
||||
}
|
||||
|
||||
#[derive(Args, Clone)]
|
||||
pub struct ConnectInventoryOpts {
|
||||
#[command(subcommand)]
|
||||
pub command: ConnectInventoryCommands,
|
||||
}
|
||||
|
||||
#[derive(Subcommand, Clone)]
|
||||
pub enum ConnectInventoryCommands {
|
||||
/// Collect the bounded inventory.environment@1 summary
|
||||
Environment(ConnectEnvironmentInventoryOpts),
|
||||
}
|
||||
|
||||
#[derive(Args, Clone)]
|
||||
pub struct ConnectEnvironmentInventoryOpts {
|
||||
/// Directory containing the persisted Connect inventory
|
||||
#[arg(long = "state-dir")]
|
||||
pub state_dir: PathBuf,
|
||||
/// Negotiated environment schema version
|
||||
#[arg(long = "schema-version", default_value_t = 1)]
|
||||
pub schema_version: u16,
|
||||
/// Negotiated environment capability
|
||||
#[arg(long, default_value = "inventory.environment@1", value_parser = NonEmptyStringValueParser::new())]
|
||||
pub capability: String,
|
||||
/// Maximum collection time in seconds
|
||||
#[arg(long = "timeout-seconds", default_value_t = 30)]
|
||||
pub timeout_seconds: u64,
|
||||
/// Confirm this explicit local L1 inventory operation
|
||||
#[arg(long = "acknowledge-l1", required = true, action = clap::ArgAction::SetTrue)]
|
||||
pub acknowledge_l1: bool,
|
||||
}
|
||||
|
||||
#[derive(Args, Clone)]
|
||||
pub struct ConnectTopOpts {
|
||||
#[command(subcommand)]
|
||||
@@ -1057,6 +1090,8 @@ pub enum CommandResult {
|
||||
ConnectRegister(ConnectRegisterOpts),
|
||||
/// Local Connect service-license command
|
||||
ConnectLicense(ConnectLicenseCommands),
|
||||
/// Explicit local Connect environment inventory command
|
||||
ConnectEnvironmentInventory(ConnectEnvironmentInventoryOpts),
|
||||
/// Consent-bound local Connect drive performance export
|
||||
ConnectDrivePerformance(ConnectDrivePerformanceOpts),
|
||||
/// Consent-bound client-to-deployment performance export
|
||||
@@ -1248,6 +1283,43 @@ mod tests {
|
||||
assert!(help.to_string().contains("Unix only"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn connect_environment_inventory_requires_explicit_l1_acknowledgement() {
|
||||
let error = Cli::try_parse_from([
|
||||
"rustfs",
|
||||
"connect",
|
||||
"inventory",
|
||||
"environment",
|
||||
"--state-dir",
|
||||
"/var/lib/rustfs/connect",
|
||||
])
|
||||
.expect_err("unacknowledged L1 inventory must fail");
|
||||
assert_eq!(error.kind(), ErrorKind::MissingRequiredArgument);
|
||||
assert!(error.to_string().contains("--acknowledge-l1"));
|
||||
|
||||
let cli = Cli::try_parse_from([
|
||||
"rustfs",
|
||||
"connect",
|
||||
"inventory",
|
||||
"environment",
|
||||
"--state-dir",
|
||||
"/var/lib/rustfs/connect",
|
||||
"--acknowledge-l1",
|
||||
])
|
||||
.expect("acknowledged environment inventory parses");
|
||||
let Some(Commands::Connect(connect)) = cli.command else {
|
||||
panic!("connect command expected");
|
||||
};
|
||||
let ConnectCommands::Inventory(inventory) = connect.command else {
|
||||
panic!("inventory command expected");
|
||||
};
|
||||
let ConnectInventoryCommands::Environment(environment) = inventory.command;
|
||||
assert_eq!(environment.schema_version, 1);
|
||||
assert_eq!(environment.capability, "inventory.environment@1");
|
||||
assert_eq!(environment.timeout_seconds, 30);
|
||||
assert!(environment.acknowledge_l1);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn connect_profile_requires_explicit_l3_acknowledgement() {
|
||||
let arguments = [
|
||||
|
||||
@@ -54,6 +54,7 @@ pub use cli::{CommandResult, InfoOpts, InfoType};
|
||||
pub use cli::{
|
||||
ConnectClientPerformanceOperation, ConnectClientPerformanceOpts, ConnectDrivePerformanceOpts, ConnectPerformanceCommands,
|
||||
};
|
||||
pub use cli::{ConnectEnvironmentInventoryOpts, ConnectInventoryCommands};
|
||||
pub use cli::{ConnectLicenseArtifactOpts, ConnectLicenseCommands, ConnectLicenseScopeOpts};
|
||||
pub use cli::{ConnectLogsMode, ConnectLogsOpts};
|
||||
pub use cli::{ConnectProfileOpts, ConnectProfileTool, ConnectThreadProfileScope};
|
||||
|
||||
@@ -19,8 +19,8 @@
|
||||
|
||||
use super::Config;
|
||||
use super::cli::{
|
||||
Cli, CommandResult, Commands, ConnectCommands, ConnectPerformanceCommands, ServerOpts, default_server_opts,
|
||||
preprocess_args_for_legacy,
|
||||
Cli, CommandResult, Commands, ConnectCommands, ConnectInventoryCommands, ConnectPerformanceCommands, ServerOpts,
|
||||
default_server_opts, preprocess_args_for_legacy,
|
||||
};
|
||||
use crate::apply_external_env_compat;
|
||||
use CommandResult::Server;
|
||||
@@ -143,6 +143,9 @@ impl Opt {
|
||||
Some(Commands::Connect(opts)) => match opts.command {
|
||||
ConnectCommands::Register(opts) => Ok(CommandResult::ConnectRegister(opts)),
|
||||
ConnectCommands::License(opts) => Ok(CommandResult::ConnectLicense(opts.command)),
|
||||
ConnectCommands::Inventory(opts) => match opts.command {
|
||||
ConnectInventoryCommands::Environment(opts) => Ok(CommandResult::ConnectEnvironmentInventory(opts)),
|
||||
},
|
||||
ConnectCommands::Performance(opts) => match opts.command {
|
||||
ConnectPerformanceCommands::Client(opts) => Ok(CommandResult::ConnectClientPerformance(*opts)),
|
||||
ConnectPerformanceCommands::Drive(opts) => Ok(CommandResult::ConnectDrivePerformance(*opts)),
|
||||
|
||||
@@ -15,9 +15,9 @@
|
||||
use crate::{
|
||||
config::{
|
||||
CommandResult, Config, ConnectClientPerformanceOperation, ConnectClientPerformanceOpts, ConnectDrivePerformanceOpts,
|
||||
ConnectLicenseCommands, ConnectLicenseScopeOpts, ConnectLogsMode, ConnectLogsOpts, ConnectProfileOpts,
|
||||
ConnectProfileTool, ConnectTelemetryArtifactOpts, ConnectTelemetryCommands, ConnectThreadProfileScope,
|
||||
ConnectTopCommands, Opt,
|
||||
ConnectEnvironmentInventoryOpts, ConnectLicenseCommands, ConnectLicenseScopeOpts, ConnectLogsMode, ConnectLogsOpts,
|
||||
ConnectProfileOpts, ConnectProfileTool, ConnectTelemetryArtifactOpts, ConnectTelemetryCommands,
|
||||
ConnectThreadProfileScope, ConnectTopCommands, Opt,
|
||||
},
|
||||
startup_lifecycle::{StartupRuntimeLifecycle, run_startup_runtime_lifecycle},
|
||||
startup_preflight::{StartupServerPreflightError, bootstrap_external_prefix_compat, init_startup_server_preflight},
|
||||
@@ -137,6 +137,7 @@ async fn async_main() -> Result<()> {
|
||||
return Ok(());
|
||||
}
|
||||
CommandResult::ConnectLicense(command) => return execute_connect_license(command),
|
||||
CommandResult::ConnectEnvironmentInventory(options) => return execute_connect_environment_inventory(options).await,
|
||||
CommandResult::ConnectClientPerformance(options) => return execute_connect_client_performance(options).await,
|
||||
CommandResult::ConnectDrivePerformance(options) => return execute_connect_drive_performance(options).await,
|
||||
CommandResult::ConnectProfile(options) => return execute_connect_profile(options).await,
|
||||
@@ -172,6 +173,41 @@ async fn async_main() -> Result<()> {
|
||||
}
|
||||
}
|
||||
|
||||
async fn execute_connect_environment_inventory(options: ConnectEnvironmentInventoryOpts) -> Result<()> {
|
||||
use crate::connect::environment::collect_environment;
|
||||
use crate::connect::inventory::InventoryStateStore;
|
||||
use crate::connect::{EnvironmentCollectionRequest, EnvironmentError};
|
||||
|
||||
let request = EnvironmentCollectionRequest::negotiate(
|
||||
options.schema_version,
|
||||
&options.capability,
|
||||
Duration::from_secs(options.timeout_seconds),
|
||||
)
|
||||
.map_err(Error::other)?;
|
||||
let store = InventoryStateStore::from_state_root(&options.state_dir).map_err(Error::other)?;
|
||||
let persisted = tokio::task::spawn_blocking(move || store.read_latest(chrono::Utc::now()))
|
||||
.await
|
||||
.map_err(Error::other)?
|
||||
.map_err(Error::other)?;
|
||||
let cancel = CancellationToken::new();
|
||||
let collection = collect_environment(&persisted.snapshot, request, &cancel);
|
||||
tokio::pin!(collection);
|
||||
let inventory = tokio::select! {
|
||||
biased;
|
||||
signal = tokio::signal::ctrl_c() => {
|
||||
signal.map_err(Error::other)?;
|
||||
cancel.cancel();
|
||||
collection.await.map_err(Error::other)?
|
||||
}
|
||||
result = collection.as_mut() => result.map_err(|error| match error {
|
||||
EnvironmentError::Cancelled => Error::other("inventory environment collection cancelled"),
|
||||
error => Error::other(error),
|
||||
})?,
|
||||
};
|
||||
println!("{}", serde_json::to_string(&inventory).map_err(Error::other)?);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn execute_connect_logs(options: ConnectLogsOpts) -> Result<()> {
|
||||
use crate::connect::{
|
||||
CaptureMode, IdentityStore, LocalLogConsent, LogCaptureRequest, LogProvenance, export_logs, save_signed_log_export,
|
||||
|
||||
@@ -14,6 +14,17 @@
|
||||
|
||||
use std::time::Duration;
|
||||
|
||||
#[cfg(target_os = "linux")]
|
||||
use std::fs;
|
||||
#[cfg(target_os = "linux")]
|
||||
use std::os::unix::fs::PermissionsExt as _;
|
||||
#[cfg(target_os = "linux")]
|
||||
use std::process::Command;
|
||||
|
||||
#[cfg(target_os = "linux")]
|
||||
use rustfs::connect::{
|
||||
CredentialStore, HeartbeatConfig, IdentityStore, InventorySchedule, InventoryStatus, spawn_inventory_runtime,
|
||||
};
|
||||
use rustfs::connect::{
|
||||
ENVIRONMENT_CAPABILITY, ENVIRONMENT_SCHEMA_VERSION, EnvironmentCollectionRequest, EnvironmentError,
|
||||
EnvironmentFilesystemType, InventorySnapshot, MAX_ENVIRONMENT_DURATION, collect_environment,
|
||||
@@ -99,3 +110,75 @@ async fn environment_collection_emits_only_the_closed_identifier_free_schema() {
|
||||
println!("inventory.environment actual output: {encoded}");
|
||||
assert!(matches!(serialized, Value::Object(_)));
|
||||
}
|
||||
|
||||
#[cfg(target_os = "linux")]
|
||||
#[tokio::test]
|
||||
async fn production_binary_collects_environment_from_persisted_inventory() {
|
||||
let temp = tempfile::tempdir_in(env!("CARGO_MANIFEST_DIR")).expect("safe tempdir");
|
||||
let state = temp.path().join("state");
|
||||
fs::create_dir(&state).expect("state root");
|
||||
fs::set_permissions(&state, fs::Permissions::from_mode(0o700)).expect("private state root");
|
||||
let config = HeartbeatConfig::new(
|
||||
"",
|
||||
Vec::new(),
|
||||
IdentityStore::new(state.join("identity")),
|
||||
CredentialStore::new(state.join("credential")),
|
||||
state.join("heartbeat/state.json"),
|
||||
);
|
||||
let shutdown = CancellationToken::new();
|
||||
let runtime = spawn_inventory_runtime(Some(config), InventorySchedule::default(), &shutdown, || {
|
||||
std::future::ready(Ok(inventory()))
|
||||
})
|
||||
.expect("state-only inventory runtime")
|
||||
.expect("configured inventory runtime");
|
||||
let mut status = runtime.status();
|
||||
tokio::time::timeout(Duration::from_secs(5), async {
|
||||
loop {
|
||||
if matches!(&*status.borrow(), InventoryStatus::Unchanged { .. }) {
|
||||
break;
|
||||
}
|
||||
status.changed().await.expect("inventory runtime remains active");
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("persisted inventory timeout");
|
||||
runtime.shutdown().await;
|
||||
|
||||
let output = Command::new(env!("CARGO_BIN_EXE_rustfs"))
|
||||
.args([
|
||||
"connect",
|
||||
"inventory",
|
||||
"environment",
|
||||
"--state-dir",
|
||||
state.to_str().expect("UTF-8 state path"),
|
||||
"--acknowledge-l1",
|
||||
])
|
||||
.output()
|
||||
.expect("run production RustFS binary");
|
||||
assert!(output.status.success(), "stderr: {}", String::from_utf8_lossy(&output.stderr));
|
||||
let actual: Value = serde_json::from_slice(&output.stdout).expect("environment JSON output");
|
||||
assert_eq!(actual["nodeCount"], 2);
|
||||
assert_eq!(actual["driveCount"], 8);
|
||||
assert_eq!(actual.as_object().expect("environment object").len(), 4);
|
||||
|
||||
for (argument, value, expected) in [
|
||||
("--schema-version", "0", "inventory_environment_unsupported_version"),
|
||||
("--capability", "inventory.environment@2", "inventory_environment_unsupported_capability"),
|
||||
] {
|
||||
let rejected = Command::new(env!("CARGO_BIN_EXE_rustfs"))
|
||||
.args([
|
||||
"connect",
|
||||
"inventory",
|
||||
"environment",
|
||||
"--state-dir",
|
||||
state.to_str().expect("UTF-8 state path"),
|
||||
"--acknowledge-l1",
|
||||
argument,
|
||||
value,
|
||||
])
|
||||
.output()
|
||||
.expect("run incompatible production command");
|
||||
assert!(!rejected.status.success());
|
||||
assert!(String::from_utf8_lossy(&rejected.stderr).contains(expected));
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user