Github-Mirror/seaweedfs
1
0 Fork 0
mirror of https://github.com/seaweedfs/seaweedfs.git synced 2026-10-20 04:54:24 +02:00
Code Issues Packages Projects Releases Wiki Activity
Files
3bd218e030d0ff220bddc164124a130ef349304a
seaweedfs/seaweed-volume/src/server/heartbeat.rs
T
Chris Lu 3bd218e030 volume: cut idle memory at high volume counts (#10861) 
* volume: start a volume's batch write worker on first use

Mounting a volume started a goroutine parked on a 128-slot channel, plus
the 128-entry batch slice it had already allocated. That is around 6.7KB
per volume the server pays whether or not the volume ever takes a write:
7231 bytes per mounted volume, of which 4101 is goroutine stack.

Only a write that asks for fsync ever reaches the worker, and a
remote-tiered or read-only volume never can. Create the channel and its
goroutine on the first such request instead, and let a write arriving
after Destroy fall back to the inline path rather than queue onto a
worker that has gone.

Measured over 20000 mounted volumes: 7231 -> 1269 bytes each.

* volume: update the heartbeat report state in place

Every heartbeat built a second map of what it was about to tell the
master, holding a freshly allocated short information message per volume,
then swapped it in over the old one -- and computed departures through a
third map of the live volume ids. A server holding 2M volumes rebuilt all
three every VolumePulsePeriod for a report that usually says nothing.

Number the heartbeats instead and mark the entry already held with the
pass that found the copy, so a quiet volume costs a map lookup and no
allocation. Departures are the entries a pass did not mark; the live-id
map is now built only when there are some, sized to them.

Measured over 10000 mounted volumes: 436 -> 196 bytes allocated per
volume per heartbeat.

* volume: fill one volume information message per heartbeat, not per volume

The heartbeat built a message for every volume held so it could hash it,
then dropped all but the few it had something to say about. At 2M volumes
that is 2M messages allocated every VolumePulsePeriod to send almost none
of them.

Fill a message the caller supplies instead, and replace it only when the
heartbeat keeps it, so a server with nothing to report fills the same one
all the way through.

Measured over 10000 mounted volumes: 196 -> 4 bytes allocated per volume
per heartbeat, and a heartbeat runs a third faster.

* volume: drop the per-volume trace from the heartbeat's status read

glog.V(4).Infof evaluates its arguments whether or not the verbosity is
on, so every volume boxed its id into a fresh interface slice on every
heartbeat: 759 of the 773 allocations a 1000-volume heartbeat made, for a
line that at this scale would print millions of unreadable rows.

Measured over 1000 mounted volumes: 4776 -> 1792 bytes and 759 -> 14
allocations per heartbeat, which no longer grows with the volume count.

* seaweed-volume: mirror the in-place heartbeat report state

Same change as the Go volume server: number the heartbeats and mark the
entry already held with the pass that found the copy, instead of building
a second map of hashes and swapping it in.

The volume snapshot must leave the reporting state as it found it, so it
keeps asking through changed() while a real heartbeat marks through
record().

* volume: refuse writes to a closed volume instead of dereferencing nil

Close and Destroy leave the needle map and data backend nil, but a caller
that already holds the volume can still reach the write path, where both
are used unguarded: a write racing a volume deletion took the server down.
syncDelete has always checked; syncWrite and the batch worker had not.

Reachable before this series and now also from the inline fallback a
durable write takes when the worker has gone.

* seaweed-volume: guard the report state with one mutex, as Go does

The full-list flag and the generation that answers it have to move
together. Split across separate atomics they cannot: a request landing
between begin's two reads returns full == false with the generation it
just raised, and one landing between commit's read and its clear is
marked answered by a heartbeat that carried no list. Either way the
resend is dropped.

Neither is reachable today -- every caller reaches this through the
store's RwLock, the flag setters under a read lock and the heartbeat
build under a write lock, so they cannot interleave. The type should not
depend on that being true two files away, and Go holds a single mutex
over exactly these fields.

* test: build the servers under test to match the harness's offset size

The mixed Go/Rust suites run both servers against one dataset, so both
have to agree on the offset width. They did not: the harness built Go
with no tags, 4-byte offsets, while the Rust crate defaults to its 5bytes
feature, and the Rust server then refused the .vif the Go server had just
written -- "bytes_offset mismatch: found 4, expected 5".

Build each side to match the offset size the test binary itself was
compiled with, so a plain `go test` and one with -tags 5BytesOffset both
get a matched pair.
2026-08-21 13:04:56 -07:00

2025 lines
75 KiB
Rust
Raw Blame History

//! Heartbeat client: registers the volume server with the master.
//!
//! Implements the bidirectional streaming `SendHeartbeat` RPC to the master,
//! matching Go's `server/volume_grpc_client_to_master.go`.
use std::collections::HashMap;
use std::path::Path;
use std::sync::atomic::Ordering;
use std::sync::Arc;
use std::time::{Duration, SystemTime, UNIX_EPOCH};
use tokio::sync::broadcast;
use tracing::{error, info, warn};
use super::grpc_client::{build_grpc_endpoint, GRPC_MAX_MESSAGE_SIZE};
use super::volume_server::VolumeServerState;
use crate::pb::master_pb;
use crate::pb::master_pb::seaweed_client::SeaweedClient;
use crate::pb::volume_server_pb;
use crate::remote_storage::s3_tier::{S3TierBackend, S3TierConfig};
use crate::storage::store::Store;
use crate::storage::volume_report::VolumeReportKey;
use crate::storage::volume_report_hash::report_hash;
use crate::storage::types::NeedleId;
const DUPLICATE_UUID_RETRY_MESSAGE: &str = "duplicate UUIDs detected, retrying connection";
const MAX_DUPLICATE_UUID_RETRIES: u32 = 3;
/// Configuration for the heartbeat client.
pub struct HeartbeatConfig {
pub ip: String,
pub port: u16,
pub grpc_port: u16,
pub public_url: String,
pub data_center: String,
pub rack: String,
pub master_addresses: Vec<String>,
pub pulse_seconds: u64,
}
/// Run the heartbeat loop using VolumeServerState.
///
/// Mirrors Go's `volume_grpc_client_to_master.go` heartbeat():
/// - On leader redirect: sleep 3s, then connect directly to the new leader
/// - On duplicate UUID error: exponential backoff (2s, 4s, 8s), exit after 3 retries
/// - On other errors: sleep pulse interval, reset to seed master list iteration
pub async fn run_heartbeat_with_state(
config: HeartbeatConfig,
state: Arc<VolumeServerState>,
mut shutdown_rx: broadcast::Receiver<()>,
) {
info!(
"Starting heartbeat to master nodes: {:?}",
config.master_addresses
);
let pulse = Duration::from_secs(config.pulse_seconds.max(1));
let mut new_leader: Option<String> = None;
let mut duplicate_retry_count: u32 = 0;
loop {
for master_addr in &config.master_addresses {
if is_stopping(&state) {
state.is_heartbeating.store(false, Ordering::Relaxed);
info!("Heartbeat stopping");
return;
}
if shutdown_rx.try_recv().is_ok() {
state.is_heartbeating.store(false, Ordering::Relaxed);
info!("Heartbeat shutting down");
return;
}
// If we have a leader redirect, sleep 3s then connect to the leader
// instead of iterating through the seed list
let target_addr = if let Some(ref leader) = new_leader {
tokio::time::sleep(Duration::from_secs(3)).await;
leader.clone()
} else {
master_addr.clone()
};
let grpc_addr = to_grpc_address(&target_addr);
info!("Connecting heartbeat to master {}", grpc_addr);
// Determine what action to take after the heartbeat attempt.
// We convert the error to a string immediately so the non-Send
// Box<dyn Error> is dropped before any .await point.
enum PostAction {
LeaderRedirect(String),
Done,
SleepDuplicate(Duration),
SleepPulse,
}
let action = match do_heartbeat(&config, &state, &grpc_addr, &target_addr, pulse, &mut shutdown_rx)
.await
{
Ok(Some(leader)) => {
info!("Master leader changed to {}", leader);
PostAction::LeaderRedirect(leader)
}
Ok(None) => {
duplicate_retry_count = 0;
PostAction::Done
}
Err(e) => {
let err_msg = e.to_string();
// Drop `e` (non-Send) before any .await
drop(e);
warn!("Heartbeat to {} error: {}", grpc_addr, err_msg);
if err_msg.contains(DUPLICATE_UUID_RETRY_MESSAGE) {
if duplicate_retry_count >= MAX_DUPLICATE_UUID_RETRIES {
error!("Shut down Volume Server due to persistent duplicate volume directories after 3 retries");
error!(
"Please check if another volume server is using the same directory"
);
std::process::exit(1);
}
let retry_delay = duplicate_uuid_retry_delay(duplicate_retry_count);
duplicate_retry_count += 1;
warn!(
"Waiting {:?} before retrying due to duplicate UUID detection (attempt {}/3)...",
retry_delay, duplicate_retry_count
);
PostAction::SleepDuplicate(retry_delay)
} else {
duplicate_retry_count = 0;
PostAction::SleepPulse
}
}
};
match action {
PostAction::LeaderRedirect(leader) => {
new_leader = Some(leader);
break;
}
PostAction::Done => {
new_leader = None;
}
PostAction::SleepDuplicate(delay) => {
new_leader = None;
tokio::time::sleep(delay).await;
}
PostAction::SleepPulse => {
new_leader = None;
tokio::time::sleep(pulse).await;
}
}
// If we connected to a leader (not seed list), break out after one attempt
// so we either reconnect to the new leader or fall back to seed list
if new_leader.is_some() {
break;
}
}
// If we have a leader redirect, skip the sleep and reconnect immediately
if new_leader.is_some() {
continue;
}
tokio::select! {
_ = tokio::time::sleep(pulse) => {}
_ = shutdown_rx.recv() => {
state.is_heartbeating.store(false, Ordering::Relaxed);
info!("Heartbeat shutting down");
return;
}
}
}
}
/// Convert a master address to a gRPC `host:port` target.
///
/// Mirrors Go's `pb.ServerToGrpcAddress()`:
/// - `host:port.grpcPort` returns `host:grpcPort` (explicit gRPC port).
/// - `host:port` returns `host:(port+10000)` (Go's default offset).
/// - Anything that fails to parse is returned unchanged.
pub fn to_grpc_address(master_addr: &str) -> String {
if let Some((host, port_str)) = master_addr.rsplit_once(':') {
// "host:port.grpcPort" — the part after the last '.' is the gRPC port.
if let Some((_, grpc_port)) = port_str.rsplit_once('.') {
if grpc_port.parse::<u16>().is_ok() {
return format!("{}:{}", host, grpc_port);
}
}
if let Ok(port) = port_str.parse::<u16>() {
let grpc_port = port + 10000;
return format!("{}:{}", host, grpc_port);
}
}
master_addr.to_string()
}
/// Call GetMasterConfiguration on seed masters before starting the heartbeat loop.
/// Mirrors Go's `checkWithMaster()` in `volume_grpc_client_to_master.go`.
/// Retries across all seed masters with a 1790ms sleep between rounds (matching Go).
/// Stores metrics address/interval from the response into server state.
async fn check_with_master(config: &HeartbeatConfig, state: &Arc<VolumeServerState>) {
loop {
for master_addr in &config.master_addresses {
let grpc_addr = to_grpc_address(master_addr);
match try_get_master_configuration(&grpc_addr, state.outgoing_grpc_tls.as_ref()).await {
Ok(resp) => {
let changed = apply_metrics_push_settings(
state,
&resp.metrics_address,
resp.metrics_interval_seconds,
);
if changed {
state.metrics_notify.notify_waiters();
}
apply_storage_backends(state, &resp.storage_backends);
info!(
"Got master configuration from {}: metrics_address={}, metrics_interval={}s",
master_addr, resp.metrics_address, resp.metrics_interval_seconds
);
return;
}
Err(e) => {
warn!("checkWithMaster {}: {}", master_addr, e);
}
}
}
tokio::time::sleep(Duration::from_millis(1790)).await;
}
}
pub async fn prime_master_configuration(config: &HeartbeatConfig, state: &Arc<VolumeServerState>) {
check_with_master(config, state).await;
}
pub async fn try_get_master_configuration(
grpc_addr: &str,
tls: Option<&super::grpc_client::OutgoingGrpcTlsConfig>,
) -> Result<master_pb::GetMasterConfigurationResponse, Box<dyn std::error::Error>> {
let channel = build_grpc_endpoint(grpc_addr, tls)?
.connect_timeout(Duration::from_secs(5))
.timeout(Duration::from_secs(10))
.connect()
.await?;
let mut client = SeaweedClient::with_interceptor(
channel,
super::request_id::outgoing_request_id_interceptor,
)
.max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE)
.max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE);
let resp = client
.get_master_configuration(master_pb::GetMasterConfigurationRequest {})
.await?;
Ok(resp.into_inner())
}
fn is_stopping(state: &VolumeServerState) -> bool {
*state.is_stopping.read().unwrap()
}
fn duplicate_uuid_retry_delay(retry_count: u32) -> Duration {
Duration::from_secs((1u64 << retry_count) * 2)
}
fn duplicate_directories(store: &Store, duplicated_uuids: &[String]) -> Vec<String> {
let mut duplicate_dirs = Vec::new();
for loc in &store.locations {
if duplicated_uuids
.iter()
.any(|uuid| uuid == &loc.directory_uuid)
{
duplicate_dirs.push(loc.directory.clone());
}
}
duplicate_dirs
}
fn apply_master_volume_options(store: &Store, hb_resp: &master_pb::HeartbeatResponse) -> bool {
if hb_resp.volume_digest_supported {
store.volume_report.accept_deltas();
}
if hb_resp.resend_full_volume_list {
info!("master asked for the full volume list");
store.volume_report.request_full_list();
}
let mut volume_opts_changed = false;
if store.get_preallocate() != hb_resp.preallocate {
store.set_preallocate(hb_resp.preallocate);
volume_opts_changed = true;
}
if hb_resp.volume_size_limit > 0
&& store.volume_size_limit.load(Ordering::Relaxed) != hb_resp.volume_size_limit
{
store
.volume_size_limit
.store(hb_resp.volume_size_limit, Ordering::Relaxed);
volume_opts_changed = true;
}
volume_opts_changed && store.maybe_adjust_volume_max()
}
type EcShardDeltaKey = (u32, String, u32, u32);
fn collect_ec_shard_delta_messages(
store: &Store,
) -> HashMap<EcShardDeltaKey, master_pb::VolumeEcShardInformationMessage> {
let mut messages = HashMap::new();
for (disk_id, loc) in store.locations.iter().enumerate() {
for (_, ec_vol) in loc.ec_volumes() {
for shard in ec_vol.shards.iter().flatten() {
messages.insert(
(
ec_vol.volume_id.0,
ec_vol.collection.clone(),
disk_id as u32,
shard.shard_id as u32,
),
master_pb::VolumeEcShardInformationMessage {
id: ec_vol.volume_id.0,
collection: ec_vol.collection.clone(),
ec_index_bits: 1u32 << shard.shard_id,
shard_sizes: vec![shard.file_size()],
disk_type: ec_vol.disk_type.to_string(),
expire_at_sec: ec_vol.expire_at_sec,
disk_id: disk_id as u32,
encode_ts_ns: ec_vol.encode_ts_ns,
..Default::default()
},
);
}
}
}
messages
}
fn diff_ec_shard_delta_messages(
previous: &HashMap<EcShardDeltaKey, master_pb::VolumeEcShardInformationMessage>,
current: &HashMap<EcShardDeltaKey, master_pb::VolumeEcShardInformationMessage>,
) -> (
Vec<master_pb::VolumeEcShardInformationMessage>,
Vec<master_pb::VolumeEcShardInformationMessage>,
) {
let mut new_ec_shards = Vec::new();
let mut deleted_ec_shards = Vec::new();
for (key, message) in current {
if previous.get(key) != Some(message) {
new_ec_shards.push(message.clone());
}
}
for (key, message) in previous {
if !current.contains_key(key) {
let mut deleted = message.clone();
deleted.shard_sizes = vec![0];
tracing::info!(
volume_id = deleted.id,
disk_id = deleted.disk_id,
ec_index_bits = deleted.ec_index_bits,
"deletes ec shards"
);
deleted_ec_shards.push(deleted);
}
}
(new_ec_shards, deleted_ec_shards)
}
/// Perform one heartbeat session with a master server.
async fn do_heartbeat(
config: &HeartbeatConfig,
state: &Arc<VolumeServerState>,
grpc_addr: &str,
current_master: &str,
pulse: Duration,
shutdown_rx: &mut broadcast::Receiver<()>,
) -> Result<Option<String>, Box<dyn std::error::Error>> {
let channel = build_grpc_endpoint(grpc_addr, state.outgoing_grpc_tls.as_ref())?
.connect_timeout(Duration::from_secs(5))
.timeout(Duration::from_secs(30))
.connect()
.await?;
let mut client = SeaweedClient::with_interceptor(
channel,
super::request_id::outgoing_request_id_interceptor,
)
.max_decoding_message_size(GRPC_MAX_MESSAGE_SIZE)
.max_encoding_message_size(GRPC_MAX_MESSAGE_SIZE);
let (tx, rx) = tokio::sync::mpsc::channel::<master_pb::Heartbeat>(32);
// This master may know nothing about this server, and has not yet said
// whether it understands digests, so start from the whole list.
state.store.read().unwrap().volume_report.reset();
// Keep track of what we sent, to generate delta updates
let (initial_hb, initial_volumes) = collect_heartbeat_with_snapshot(config, state);
let mut last_volumes: HashMap<u32, master_pb::VolumeInformationMessage> =
initial_volumes.iter().map(|v| (v.id, v.clone())).collect();
let mut last_ec_shards = {
let store = state.store.read().unwrap();
collect_ec_shard_delta_messages(&store)
};
// Send initial heartbeats BEFORE calling send_heartbeat to avoid deadlock:
// the server won't send response headers until it receives the first message,
// but send_heartbeat().await waits for response headers.
tx.send(initial_hb).await?;
tx.send(collect_ec_heartbeat(config, state)).await?;
let stream = tokio_stream::wrappers::ReceiverStream::new(rx);
let mut response_stream = client.send_heartbeat(stream).await?.into_inner();
info!("Heartbeat stream established with {}", grpc_addr);
// Publish the master we're now talking to in canonical http host:port
// form so Ping admission can recognise it once a leader change moves us
// off the seed list. Mirrors Go's vs.setCurrentMaster(masterAddress).
{
let normalised =
super::volume_server::to_http_address(current_master).into_owned();
let mut guard = state.current_master_url.write().await;
*guard = normalised;
}
if is_stopping(state) {
state.is_heartbeating.store(false, Ordering::Relaxed);
send_deregister_heartbeat(config, state, &tx).await;
info!("Heartbeat stopping");
return Ok(None);
}
state.is_heartbeating.store(true, Ordering::Relaxed);
let mut volume_tick = tokio::time::interval(pulse);
let mut ec_tick = tokio::time::interval(pulse * 17);
volume_tick.tick().await;
ec_tick.tick().await;
loop {
tokio::select! {
resp = response_stream.message() => {
match resp {
Ok(Some(hb_resp)) => {
// Match Go ordering: DuplicatedUuids first, then volume
// options, then leader redirect.
if !hb_resp.duplicated_uuids.is_empty() {
let duplicate_dirs = {
let store = state.store.read().unwrap();
duplicate_directories(&store, &hb_resp.duplicated_uuids)
};
error!(
"Master reported duplicate volume directories: {:?}",
duplicate_dirs
);
return Err(format!(
"{}: {:?}",
DUPLICATE_UUID_RETRY_MESSAGE, duplicate_dirs
)
.into());
}
let changed = {
let s = state.store.read().unwrap();
apply_master_volume_options(&s, &hb_resp)
};
if changed {
let (adjusted_hb, adjusted_volumes) =
collect_heartbeat_with_snapshot(config, state);
last_volumes =
adjusted_volumes.iter().map(|v| (v.id, v.clone())).collect();
last_ec_shards = {
let store = state.store.read().unwrap();
collect_ec_shard_delta_messages(&store)
};
if tx.send(adjusted_hb).await.is_err() {
return Ok(None);
}
}
let metrics_changed = apply_metrics_push_settings(
state,
&hb_resp.metrics_address,
hb_resp.metrics_interval_seconds,
);
if metrics_changed {
state.metrics_notify.notify_waiters();
}
// Match Go: only redirect if leader is non-empty AND
// different from the current master we're connected to.
if !hb_resp.leader.is_empty() && current_master != hb_resp.leader {
return Ok(Some(hb_resp.leader));
}
}
Ok(None) => return Ok(None),
Err(e) => return Err(Box::new(e)),
}
}
_ = volume_tick.tick() => {
{
let s = state.store.read().unwrap();
s.maybe_adjust_volume_max();
}
let (current_hb, current_volumes) = collect_heartbeat_with_snapshot(config, state);
last_volumes = current_volumes.iter().map(|v| (v.id, v.clone())).collect();
last_ec_shards = {
let store = state.store.read().unwrap();
collect_ec_shard_delta_messages(&store)
};
if tx.send(current_hb).await.is_err() {
return Ok(None);
}
}
_ = ec_tick.tick() => {
let current_ec_hb = collect_ec_heartbeat(config, state);
last_ec_shards = {
let store = state.store.read().unwrap();
collect_ec_shard_delta_messages(&store)
};
if tx.send(current_ec_hb).await.is_err() {
return Ok(None);
}
}
_ = state.volume_state_notify.notified() => {
if is_stopping(state) {
state.is_heartbeating.store(false, Ordering::Relaxed);
send_deregister_heartbeat(config, state, &tx).await;
info!("Heartbeat stopping");
return Ok(None);
}
let held_volumes = collect_volume_snapshot(config, state);
let current_volumes: HashMap<u32, _> = held_volumes.iter().map(|v| (v.id, v.clone())).collect();
let current_ec_shards = {
let store = state.store.read().unwrap();
collect_ec_shard_delta_messages(&store)
};
let mut new_vols = Vec::new();
let mut del_vols = Vec::new();
for (id, vol) in &current_volumes {
if !last_volumes.contains_key(id) {
new_vols.push(master_pb::VolumeShortInformationMessage {
id: *id,
collection: vol.collection.clone(),
version: vol.version,
replica_placement: vol.replica_placement,
ttl: vol.ttl,
disk_type: vol.disk_type.clone(),
disk_id: vol.disk_id,
});
}
}
for (id, vol) in &last_volumes {
if !current_volumes.contains_key(id) {
del_vols.push(master_pb::VolumeShortInformationMessage {
id: *id,
collection: vol.collection.clone(),
version: vol.version,
replica_placement: vol.replica_placement,
ttl: vol.ttl,
disk_type: vol.disk_type.clone(),
disk_id: vol.disk_id,
});
}
}
let (new_ec_shards, deleted_ec_shards) =
diff_ec_shard_delta_messages(&last_ec_shards, &current_ec_shards);
// Collect current state for state-only or combined delta heartbeats.
// Mirrors Go's StateUpdateChan case which sends state changes immediately.
let current_state = Some(volume_server_pb::VolumeServerState {
maintenance: state.maintenance.load(Ordering::Relaxed),
version: state.state_version.load(Ordering::Relaxed),
});
if !new_vols.is_empty()
|| !del_vols.is_empty()
|| !new_ec_shards.is_empty()
|| !deleted_ec_shards.is_empty()
{
let delta_hb = master_pb::Heartbeat {
ip: config.ip.clone(),
port: config.port as u32,
grpc_port: config.grpc_port as u32,
public_url: config.public_url.clone(),
data_center: config.data_center.clone(),
rack: config.rack.clone(),
new_volumes: new_vols,
deleted_volumes: del_vols,
new_ec_shards,
deleted_ec_shards,
state: current_state,
..Default::default()
};
if tx.send(delta_hb).await.is_err() {
return Ok(None);
}
last_volumes = current_volumes;
last_ec_shards = current_ec_shards;
} else {
// State-only heartbeat (e.g., MarkReadonly/MarkWritable changed state
// without adding/removing volumes). Mirrors Go's StateUpdateChan case.
let state_hb = master_pb::Heartbeat {
ip: config.ip.clone(),
port: config.port as u32,
grpc_port: config.grpc_port as u32,
data_center: config.data_center.clone(),
rack: config.rack.clone(),
state: current_state,
..Default::default()
};
if tx.send(state_hb).await.is_err() {
return Ok(None);
}
}
}
_ = shutdown_rx.recv() => {
state.is_heartbeating.store(false, Ordering::Relaxed);
send_deregister_heartbeat(config, state, &tx).await;
info!("Sent deregistration heartbeat");
return Ok(None);
}
}
}
}
async fn send_deregister_heartbeat(
config: &HeartbeatConfig,
state: &Arc<VolumeServerState>,
tx: &tokio::sync::mpsc::Sender<master_pb::Heartbeat>,
) {
let empty = {
let store = state.store.read().unwrap();
// Deregister: no effective max computed, fall back to configured max.
let (location_uuids, disk_tags) = collect_location_metadata(&store, &[]);
master_pb::Heartbeat {
id: store.id.clone(),
ip: config.ip.clone(),
port: config.port as u32,
public_url: config.public_url.clone(),
max_file_key: 0,
data_center: config.data_center.clone(),
rack: config.rack.clone(),
has_no_volumes: true,
has_no_ec_shards: true,
grpc_port: config.grpc_port as u32,
location_uuids,
disk_tags,
..Default::default()
}
};
let _ = tx.send(empty).await;
tokio::time::sleep(Duration::from_millis(200)).await;
}
fn apply_metrics_push_settings(
state: &VolumeServerState,
address: &str,
interval_seconds: u32,
) -> bool {
let mut runtime = state.metrics_runtime.write().unwrap();
if runtime.push_gateway.address == address
&& runtime.push_gateway.interval_seconds == interval_seconds
{
return false;
}
runtime.push_gateway.address = address.to_string();
runtime.push_gateway.interval_seconds = interval_seconds;
true
}
fn apply_storage_backends(
state: &VolumeServerState,
storage_backends: &[master_pb::StorageBackend],
) {
if storage_backends.is_empty() {
return;
}
let mut registry = state.s3_tier_registry.write().unwrap();
let mut global_registry = crate::remote_storage::s3_tier::global_s3_tier_registry()
.write()
.unwrap();
for backend in storage_backends {
if backend.r#type != "s3" {
continue;
}
let properties = &backend.properties;
let config = S3TierConfig {
access_key: properties
.get("aws_access_key_id")
.cloned()
.unwrap_or_default(),
secret_key: properties
.get("aws_secret_access_key")
.cloned()
.unwrap_or_default(),
region: properties.get("region").cloned().unwrap_or_default(),
bucket: properties.get("bucket").cloned().unwrap_or_default(),
endpoint: properties.get("endpoint").cloned().unwrap_or_default(),
storage_class: properties.get("storage_class").cloned().unwrap_or_default(),
force_path_style: parse_bool_property(properties.get("force_path_style")),
};
let backend_id = if backend.id.is_empty() {
"default"
} else {
backend.id.as_str()
};
register_s3_backend(&mut registry, backend, backend_id, &config);
register_s3_backend(&mut global_registry, backend, backend_id, &config);
}
}
fn register_s3_backend(
registry: &mut crate::remote_storage::s3_tier::S3TierRegistry,
backend: &master_pb::StorageBackend,
backend_id: &str,
config: &S3TierConfig,
) {
let qualified_name = format!("{}.{}", backend.r#type, backend_id);
if registry.get(&qualified_name).is_none() {
registry.register(qualified_name, S3TierBackend::new(config));
}
if backend_id == "default" && registry.get(&backend.r#type).is_none() {
registry.register(backend.r#type.clone(), S3TierBackend::new(config));
}
}
fn parse_bool_property(value: Option<&String>) -> bool {
value
.map(|v| {
matches!(
v.trim().to_ascii_lowercase().as_str(),
"1" | "t" | "true" | "y" | "yes" | "on"
)
})
.unwrap_or(true)
}
/// Collect volume information into a Heartbeat message.
fn collect_heartbeat_with_snapshot(
config: &HeartbeatConfig,
state: &Arc<VolumeServerState>,
) -> (master_pb::Heartbeat, Vec<master_pb::VolumeInformationMessage>) {
let mut store = state.store.write().unwrap();
let (ec_shards, deleted_ec_shards) = store.delete_expired_ec_volumes();
build_heartbeat_with_ec_status(
config,
&mut store,
deleted_ec_shards,
ec_shards.is_empty(),
true,
)
}
/// Lists the volumes the server holds without touching reporting state or
/// expiring anything, for callers that only need to diff against a previous
/// snapshot and send a message of their own.
fn collect_volume_snapshot(
config: &HeartbeatConfig,
state: &Arc<VolumeServerState>,
) -> Vec<master_pb::VolumeInformationMessage> {
let mut store = state.store.write().unwrap();
build_heartbeat_with_ec_status(config, &mut store, Vec::new(), true, false).1
}
/// The heartbeat alone, without the volume snapshot the send loop pairs it
/// with. Only the tests want it that way; the loop calls
/// collect_heartbeat_with_snapshot directly.
#[cfg(test)]
fn collect_heartbeat(
config: &HeartbeatConfig,
state: &Arc<VolumeServerState>,
) -> master_pb::Heartbeat {
collect_heartbeat_with_snapshot(config, state).0
}
fn collect_location_metadata(
store: &Store,
disk_max_by_id: &[i32],
) -> (Vec<String>, Vec<master_pb::DiskTag>) {
let location_uuids = store
.locations
.iter()
.map(|loc| loc.directory_uuid.clone())
.collect();
let disk_tags = store
.locations
.iter()
.enumerate()
.map(|(disk_id, loc)| {
let max_volume_count = disk_max_by_id
.get(disk_id)
.copied()
.unwrap_or_else(|| loc.max_volume_count.load(Ordering::Relaxed));
master_pb::DiskTag {
disk_id: disk_id as u32,
tags: loc.tags.clone(),
r#type: loc.disk_type.to_string(),
max_volume_count: max_volume_count as i64,
}
})
.collect();
(location_uuids, disk_tags)
}
#[cfg(test)]
fn build_heartbeat(config: &HeartbeatConfig, store: &mut Store) -> master_pb::Heartbeat {
let has_no_ec_shards = collect_live_ec_shards(store, false).is_empty();
build_heartbeat_with_ec_status(config, store, Vec::new(), has_no_ec_shards, true).0
}
/// Returns the heartbeat to send and, separately, every volume held. The
/// caller derives mount and unmount deltas by diffing successive snapshots, so
/// it must not be handed the partial list a heartbeat may carry.
fn build_heartbeat_with_ec_status(
config: &HeartbeatConfig,
store: &mut Store,
deleted_ec_shards: Vec<master_pb::VolumeEcShardInformationMessage>,
has_no_ec_shards: bool,
commit_report: bool,
) -> (master_pb::Heartbeat, Vec<master_pb::VolumeInformationMessage>) {
const MAX_TTL_VOLUME_REMOVAL_DELAY: u32 = 10;
#[derive(Default)]
struct ReadOnlyCounts {
is_read_only: u32,
no_write_or_delete: u32,
no_write_can_delete: u32,
is_disk_space_low: u32,
}
let mut volumes = Vec::new();
// Covers every volume held, whether or not this heartbeat names it, so the
// master can tell whether applying what it was sent leaves it current.
// Volumes skipped below -- quarantined, phantom, expired -- are in neither.
let mut volume_digest: u64 = 0;
let (send_full_list, report_generation, report_pass) = store.volume_report.begin();
let mut changed_volumes = Vec::new();
let mut max_file_key = NeedleId(0);
let mut max_volume_counts: HashMap<String, u32> = HashMap::new();
let mut disk_total_bytes: HashMap<String, u64> = HashMap::new();
let mut disk_free_bytes: HashMap<String, u64> = HashMap::new();
// Collect per-collection disk size and read-only counts for metrics
let mut disk_sizes: HashMap<String, (u64, u64)> = HashMap::new(); // (normal, deleted)
let mut ro_counts: HashMap<String, ReadOnlyCounts> = HashMap::new();
let volume_size_limit = store.volume_size_limit.load(Ordering::Relaxed);
// Per-disk effective max for DiskTag, captured alongside the per-type sum.
let mut disk_max_by_id = vec![0i32; store.locations.len()];
for (disk_id, loc) in store.locations.iter_mut().enumerate() {
let disk_type_str = loc.disk_type.to_string();
let mut effective_max_count = loc.max_volume_count.load(Ordering::Relaxed);
if loc.is_disk_space_low.load(Ordering::Relaxed) {
let used_slots = loc.volumes_len() as i32
+ ((loc.ec_shard_count()
+ crate::storage::erasure_coding::ec_shard::DATA_SHARDS_COUNT
- 1)
/ crate::storage::erasure_coding::ec_shard::DATA_SHARDS_COUNT)
as i32;
effective_max_count = used_slots;
}
if effective_max_count < 0 {
effective_max_count = 0;
}
*max_volume_counts.entry(disk_type_str.clone()).or_insert(0) += effective_max_count as u32;
disk_max_by_id[disk_id] = effective_max_count;
// Sum capacity per disk type; assumes one location per filesystem. Locations
// sharing a mount over-report absolute bytes but not the used ratio the gate
// uses. Mirrors weed/storage/store.go.
*disk_total_bytes.entry(disk_type_str.clone()).or_insert(0) +=
loc.disk_total_bytes.load(Ordering::Relaxed);
*disk_free_bytes.entry(disk_type_str).or_insert(0) +=
loc.disk_free_bytes.load(Ordering::Relaxed);
let mut delete_vids = Vec::new();
for (_, vol) in loc.iter_volumes() {
let cur_max = vol.max_file_key();
if cur_max > max_file_key {
max_file_key = cur_max;
}
let volume_size = vol.dat_file_size().unwrap_or(0);
let mut should_delete_volume = false;
if vol.last_io_error().is_some() {
delete_vids.push(vol.id);
should_delete_volume = true;
} else if !vol.is_expired(volume_size, volume_size_limit) {
// Detect phantom volumes: the .dat was unlinked from disk but is still
// held open as a deleted FD, so the volume keeps serving and heartbeating
// while no disk-path operation can ever succeed. Skip remote-tiered volumes,
// whose .dat legitimately lives in cloud storage. Only a present .dat is
// cached for 30s; a missing one is re-checked every heartbeat so the volume
// stays suppressed until the file returns. See issues/10004
if vol.file_count() > 0 && !vol.has_remote_file {
const DISK_CHECK_INTERVAL_NS: i64 = 30 * 1_000_000_000;
let now_ns = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or(Duration::ZERO)
.as_nanos() as i64;
if now_ns - vol.last_disk_check_ns.load(Ordering::Relaxed) > DISK_CHECK_INTERVAL_NS {
if !Path::new(&vol.file_name(".dat")).exists() {
warn!("Volume {}: data file {} missing (held open as deleted FD) - not reporting to master", vol.id.0, vol.file_name(".dat"));
continue;
}
vol.last_disk_check_ns.store(now_ns, Ordering::Relaxed);
}
}
let (remote_storage_name, remote_storage_key) = vol.remote_storage_name_key();
let volume_message = master_pb::VolumeInformationMessage {
id: vol.id.0,
size: volume_size,
collection: vol.collection.clone(),
file_count: vol.file_count() as u64,
delete_count: vol.deleted_count() as u64,
deleted_byte_count: vol.deleted_size(),
read_only: vol.is_read_only(),
replica_placement: vol.super_block.replica_placement.to_byte() as u32,
version: vol.super_block.version.0 as u32,
ttl: vol.super_block.ttl.to_u32(),
compact_revision: vol.super_block.compaction_revision as u32,
modified_at_second: vol.last_modified_ts() as i64,
disk_type: loc.disk_type.to_string(),
disk_id: disk_id as u32,
remote_storage_name,
remote_storage_key,
};
let hash = report_hash(&volume_message);
volume_digest ^= hash;
let key: VolumeReportKey = (volume_message.disk_id, volume_message.id);
// A snapshot must leave the reporting state as it found it, so
// it asks rather than marks.
let is_news = if commit_report {
store.volume_report.record(key, hash, report_pass)
} else {
store.volume_report.changed(key, hash)
};
if send_full_list || is_news {
changed_volumes.push(volume_message.clone());
}
volumes.push(volume_message);
} else if vol.is_expired_long_enough(MAX_TTL_VOLUME_REMOVAL_DELAY) {
delete_vids.push(vol.id);
should_delete_volume = true;
}
// Track disk size by collection
let entry = disk_sizes.entry(vol.collection.clone()).or_insert((0, 0));
if !should_delete_volume {
entry.0 += volume_size;
entry.1 += vol.deleted_size();
}
let read_only = ro_counts.entry(vol.collection.clone()).or_default();
if !should_delete_volume && vol.is_read_only() {
read_only.is_read_only += 1;
if vol.is_no_write_or_delete() {
read_only.no_write_or_delete += 1;
}
if vol.is_no_write_can_delete() {
read_only.no_write_can_delete += 1;
}
if loc.is_disk_space_low.load(Ordering::Relaxed) {
read_only.is_disk_space_low += 1;
}
}
}
for vid in delete_vids {
let _ = loc.delete_volume(vid, false, false);
}
}
// Update disk size and read-only gauges
for (col, (normal, deleted)) in &disk_sizes {
crate::metrics::DISK_SIZE_GAUGE
.with_label_values(&[col, crate::metrics::DISK_SIZE_LABEL_NORMAL])
.set(*normal as f64);
crate::metrics::DISK_SIZE_GAUGE
.with_label_values(&[col, crate::metrics::DISK_SIZE_LABEL_DELETED_BYTES])
.set(*deleted as f64);
}
for (col, counts) in &ro_counts {
crate::metrics::READ_ONLY_VOLUME_GAUGE
.with_label_values(&[col, crate::metrics::READ_ONLY_LABEL_IS_READ_ONLY])
.set(counts.is_read_only as f64);
crate::metrics::READ_ONLY_VOLUME_GAUGE
.with_label_values(&[col, crate::metrics::READ_ONLY_LABEL_NO_WRITE_OR_DELETE])
.set(counts.no_write_or_delete as f64);
crate::metrics::READ_ONLY_VOLUME_GAUGE
.with_label_values(&[col, crate::metrics::READ_ONLY_LABEL_NO_WRITE_CAN_DELETE])
.set(counts.no_write_can_delete as f64);
crate::metrics::READ_ONLY_VOLUME_GAUGE
.with_label_values(&[col, crate::metrics::READ_ONLY_LABEL_IS_DISK_SPACE_LOW])
.set(counts.is_disk_space_low as f64);
}
// Update max volumes gauge
let total_max: i64 = max_volume_counts.values().map(|v| *v as i64).sum();
crate::metrics::MAX_VOLUMES.set(total_max);
// Only when this heartbeat is going to be sent: marking volumes reported
// and then discarding the message would leave the master never told.
if commit_report {
store.volume_report.commit(report_pass, report_generation);
}
// has_no_volumes says the server holds nothing, so it may only be derived
// from a full list. Deriving it from a changed-only heartbeat would make a
// quiet one read as an empty server and drop every volume on it.
let (heartbeat_volumes, changed_volumes, has_no_volumes) = if send_full_list {
(changed_volumes, Vec::new(), volumes.is_empty())
} else {
(Vec::new(), changed_volumes, false)
};
let (location_uuids, disk_tags) = collect_location_metadata(store, &disk_max_by_id);
let heartbeat = master_pb::Heartbeat {
id: store.id.clone(),
ip: config.ip.clone(),
port: config.port as u32,
public_url: config.public_url.clone(),
max_file_key: max_file_key.0,
data_center: config.data_center.clone(),
rack: config.rack.clone(),
admin_port: config.port as u32,
volumes: heartbeat_volumes,
changed_volumes,
volume_digest: Some(volume_digest),
deleted_ec_shards,
has_no_volumes,
has_no_ec_shards,
max_volume_counts,
disk_total_bytes,
disk_free_bytes,
grpc_port: config.grpc_port as u32,
location_uuids,
disk_tags,
..Default::default()
};
(heartbeat, volumes)
}
fn collect_live_ec_shards(
store: &Store,
update_metrics: bool,
) -> Vec<master_pb::VolumeEcShardInformationMessage> {
let mut ec_shards = Vec::new();
let mut ec_sizes: HashMap<String, u64> = HashMap::new();
for (disk_id, loc) in store.locations.iter().enumerate() {
for (_, ec_vol) in loc.ec_volumes() {
for message in ec_vol.to_volume_ec_shard_information_messages(disk_id as u32) {
if update_metrics {
let total_size: u64 = message
.shard_sizes
.iter()
.map(|size| (*size).max(0) as u64)
.sum();
*ec_sizes.entry(message.collection.clone()).or_insert(0) += total_size;
}
ec_shards.push(message);
}
}
}
if update_metrics {
for (col, size) in &ec_sizes {
crate::metrics::DISK_SIZE_GAUGE
.with_label_values(&[col, crate::metrics::DISK_SIZE_LABEL_EC])
.set(*size as f64);
}
}
ec_shards
}
/// Collect EC shard information into a Heartbeat message.
fn collect_ec_heartbeat(config: &HeartbeatConfig, state: &Arc<VolumeServerState>) -> master_pb::Heartbeat {
let store = state.store.read().unwrap();
let ec_shards = collect_live_ec_shards(&store, true);
let has_no = ec_shards.is_empty();
master_pb::Heartbeat {
ip: config.ip.clone(),
port: config.port as u32,
grpc_port: config.grpc_port as u32,
data_center: config.data_center.clone(),
rack: config.rack.clone(),
ec_shards,
has_no_ec_shards: has_no,
..Default::default()
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::config::MinFreeSpace;
use crate::config::ReadMode;
use crate::metrics::{
DISK_SIZE_GAUGE, DISK_SIZE_LABEL_DELETED_BYTES, DISK_SIZE_LABEL_EC,
DISK_SIZE_LABEL_NORMAL, READ_ONLY_LABEL_IS_DISK_SPACE_LOW,
READ_ONLY_LABEL_IS_READ_ONLY, READ_ONLY_LABEL_NO_WRITE_CAN_DELETE,
READ_ONLY_LABEL_NO_WRITE_OR_DELETE, READ_ONLY_VOLUME_GAUGE,
};
use crate::remote_storage::s3_tier::S3TierRegistry;
use crate::security::{Guard, SigningKey};
use crate::storage::needle_map::NeedleMapKind;
use crate::storage::types::{DiskType, Version, VolumeId};
use std::sync::atomic::Ordering;
use std::sync::RwLock;
use std::time::{SystemTime, UNIX_EPOCH};
fn test_config() -> HeartbeatConfig {
HeartbeatConfig {
ip: "127.0.0.1".to_string(),
port: 8080,
grpc_port: 18080,
public_url: "127.0.0.1:8080".to_string(),
data_center: "dc1".to_string(),
rack: "rack1".to_string(),
master_addresses: Vec::new(),
pulse_seconds: 5,
}
}
fn test_state_with_store(store: Store) -> Arc<VolumeServerState> {
Arc::new(VolumeServerState {
store: RwLock::new(store),
guard: RwLock::new(Guard::new(
&[],
SigningKey(vec![]),
0,
SigningKey(vec![]),
0,
)),
is_stopping: RwLock::new(false),
maintenance: std::sync::atomic::AtomicBool::new(false),
state_version: std::sync::atomic::AtomicU32::new(0),
concurrent_upload_limit: 0,
concurrent_download_limit: 0,
inflight_upload_data_timeout: std::time::Duration::from_secs(60),
inflight_download_data_timeout: std::time::Duration::from_secs(60),
inflight_upload_bytes: std::sync::atomic::AtomicI64::new(0),
inflight_download_bytes: std::sync::atomic::AtomicI64::new(0),
upload_notify: tokio::sync::Notify::new(),
download_notify: tokio::sync::Notify::new(),
data_center: String::new(),
rack: String::new(),
file_size_limit_bytes: 0,
maintenance_byte_per_second: 0,
is_heartbeating: std::sync::atomic::AtomicBool::new(false),
has_master: true,
pre_stop_seconds: 0,
volume_state_notify: tokio::sync::Notify::new(),
write_queue: std::sync::OnceLock::new(),
s3_tier_registry: std::sync::RwLock::new(S3TierRegistry::new()),
read_mode: ReadMode::Local,
allow_untrusted_remote_endpoints: false,
master_url: String::new(),
master_urls: Vec::new(),
seed_master_set: std::collections::HashSet::new(),
current_master_url: tokio::sync::RwLock::new(String::new()),
self_url: String::new(),
http_client: reqwest::Client::new(),
outgoing_http_scheme: "http".to_string(),
outgoing_grpc_tls: None,
metrics_runtime: std::sync::RwLock::new(Default::default()),
metrics_notify: tokio::sync::Notify::new(),
fix_jpg_orientation: false,
has_slow_read: true,
read_buffer_size_bytes: 4 * 1024 * 1024,
security_file: String::new(),
cli_white_list: vec![],
state_file_path: String::new(),
})
}
#[test]
fn test_to_grpc_address_default_offset() {
assert_eq!(to_grpc_address("10.0.0.1:9333"), "10.0.0.1:19333");
assert_eq!(to_grpc_address("localhost:9333"), "localhost:19333");
}
#[test]
fn test_to_grpc_address_explicit_grpc_port() {
// host:port.grpcPort form — gRPC port is what's after the dot.
assert_eq!(to_grpc_address("10.85.183.6:5300.6300"), "10.85.183.6:6300");
assert_eq!(to_grpc_address("master.local:9333.19333"), "master.local:19333");
}
#[test]
fn test_to_grpc_address_returns_input_when_unparseable() {
assert_eq!(to_grpc_address(""), "");
assert_eq!(to_grpc_address("no-port"), "no-port");
assert_eq!(to_grpc_address("host:not-a-port"), "host:not-a-port");
}
#[test]
fn test_build_heartbeat_includes_store_identity_and_disk_metadata() {
let temp_dir = tempfile::tempdir().unwrap();
let dir = temp_dir.path().to_str().unwrap();
let mut store = Store::new(NeedleMapKind::InMemory);
store.id = "volume-node-a".to_string();
store
.add_location(
dir,
dir,
3,
DiskType::HardDrive,
MinFreeSpace::Percent(1.0),
vec!["fast".to_string(), "ssd".to_string()],
)
.unwrap();
store
.add_volume(
VolumeId(7),
"pics",
None,
None,
0,
DiskType::HardDrive,
Version::current(),
)
.unwrap();
let heartbeat = build_heartbeat(&test_config(), &mut store);
assert_eq!(heartbeat.id, "volume-node-a");
assert_eq!(heartbeat.volumes.len(), 1);
assert!(!heartbeat.has_no_volumes);
assert_eq!(
heartbeat.location_uuids,
vec![store.locations[0].directory_uuid.clone()]
);
assert_eq!(heartbeat.disk_tags.len(), 1);
assert_eq!(heartbeat.disk_tags[0].disk_id, 0);
assert_eq!(
heartbeat.disk_tags[0].tags,
vec!["fast".to_string(), "ssd".to_string()]
);
assert_eq!(heartbeat.disk_tags[0].r#type, DiskType::HardDrive.to_string());
assert_eq!(heartbeat.disk_tags[0].max_volume_count, 3);
}
#[test]
fn test_build_heartbeat_disk_tag_reflects_disk_space_low_override() {
let temp_dir = tempfile::tempdir().unwrap();
let dir = temp_dir.path().to_str().unwrap();
let mut store = Store::new(NeedleMapKind::InMemory);
store
.add_location(
dir,
dir,
3,
DiskType::HardDrive,
MinFreeSpace::Percent(1.0),
vec![],
)
.unwrap();
store
.add_volume(
VolumeId(7),
"pics",
None,
None,
0,
DiskType::HardDrive,
Version::current(),
)
.unwrap();
// Low disk space caps the per-disk max at used slots (1 volume, 0 EC).
store.locations[0]
.is_disk_space_low
.store(true, std::sync::atomic::Ordering::Relaxed);
let heartbeat = build_heartbeat(&test_config(), &mut store);
assert_eq!(heartbeat.disk_tags[0].max_volume_count, 1);
assert_eq!(heartbeat.max_volume_counts[&DiskType::HardDrive.to_string()], 1);
}
#[test]
fn test_build_heartbeat_marks_empty_store_as_has_no_volumes() {
let temp_dir = tempfile::tempdir().unwrap();
let dir = temp_dir.path().to_str().unwrap();
let mut store = Store::new(NeedleMapKind::InMemory);
store.id = "volume-node-b".to_string();
store
.add_location(
dir,
dir,
2,
DiskType::HardDrive,
MinFreeSpace::Percent(1.0),
Vec::new(),
)
.unwrap();
let heartbeat = build_heartbeat(&test_config(), &mut store);
assert!(heartbeat.volumes.is_empty());
assert!(heartbeat.has_no_volumes);
}
// The digest must cover exactly the volumes the heartbeat carries. A volume
// reported but left out of the digest, or the reverse, makes the master's
// comparison disagree forever. An empty store still reports a digest, so
// the master can tell it from a server that computes none.
fn reporting_store(dir: &str, count: u32) -> Store {
let mut store = Store::new(NeedleMapKind::InMemory);
store
.add_location(
dir,
dir,
16,
DiskType::HardDrive,
MinFreeSpace::Percent(1.0),
Vec::new(),
)
.unwrap();
for id in 1..=count {
store
.add_volume(
VolumeId(id),
"pics",
None,
None,
0,
DiskType::HardDrive,
Version::current(),
)
.unwrap();
}
store.volume_report.reset();
store
}
// Until the master says it compares digests it may be one that reads a
// partial list as the whole truth, so it keeps getting the whole list.
#[test]
fn test_heartbeat_sends_full_list_until_the_master_accepts() {
let temp_dir = tempfile::tempdir().unwrap();
let mut store = reporting_store(temp_dir.path().to_str().unwrap(), 2);
for _ in 0..3 {
let heartbeat = build_heartbeat(&test_config(), &mut store);
assert_eq!(heartbeat.volumes.len(), 2);
assert!(heartbeat.changed_volumes.is_empty());
}
}
// The one that would be catastrophic: a heartbeat with nothing to report
// must not look like a server that has lost every volume.
#[test]
fn test_quiet_heartbeat_does_not_look_like_an_empty_server() {
let temp_dir = tempfile::tempdir().unwrap();
let mut store = reporting_store(temp_dir.path().to_str().unwrap(), 2);
store.volume_report.accept_deltas();
let full = build_heartbeat(&test_config(), &mut store);
let quiet = build_heartbeat(&test_config(), &mut store);
assert!(quiet.volumes.is_empty());
assert!(quiet.changed_volumes.is_empty());
assert!(!quiet.has_no_volumes);
// The digest still covers everything held, not just what was sent.
assert_eq!(quiet.volume_digest, full.volume_digest);
}
#[test]
fn test_heartbeat_reports_only_what_changed() {
let temp_dir = tempfile::tempdir().unwrap();
let mut store = reporting_store(temp_dir.path().to_str().unwrap(), 2);
store.volume_report.accept_deltas();
build_heartbeat(&test_config(), &mut store);
store
.add_volume(
VolumeId(3),
"pics",
None,
None,
0,
DiskType::HardDrive,
Version::current(),
)
.unwrap();
let heartbeat = build_heartbeat(&test_config(), &mut store);
assert!(heartbeat.volumes.is_empty());
assert_eq!(heartbeat.changed_volumes.len(), 1);
assert_eq!(heartbeat.changed_volumes[0].id, 3);
}
#[test]
fn test_heartbeat_returns_to_the_full_list_on_request() {
let temp_dir = tempfile::tempdir().unwrap();
let mut store = reporting_store(temp_dir.path().to_str().unwrap(), 2);
store.volume_report.accept_deltas();
build_heartbeat(&test_config(), &mut store);
store.volume_report.request_full_list();
let resent = build_heartbeat(&test_config(), &mut store);
assert_eq!(resent.volumes.len(), 2);
assert!(resent.changed_volumes.is_empty());
let next = build_heartbeat(&test_config(), &mut store);
assert!(next.volumes.is_empty());
}
// A request that lands while a heartbeat is being built asked about a later
// state than that heartbeat carries, so it must survive being committed over.
#[test]
fn test_full_list_request_during_collection_survives() {
let temp_dir = tempfile::tempdir().unwrap();
let mut store = reporting_store(temp_dir.path().to_str().unwrap(), 2);
store.volume_report.accept_deltas();
build_heartbeat(&test_config(), &mut store);
let (full, generation, pass) = store.volume_report.begin();
assert!(!full);
store.volume_report.request_full_list();
store.volume_report.commit(pass, generation);
let heartbeat = build_heartbeat(&test_config(), &mut store);
assert_eq!(heartbeat.volumes.len(), 2);
}
// Taking a snapshot must not mark volumes as told to a master that is
// getting a different message.
#[test]
fn test_snapshot_does_not_mark_volumes_reported() {
let temp_dir = tempfile::tempdir().unwrap();
let mut store = reporting_store(temp_dir.path().to_str().unwrap(), 2);
store.volume_report.accept_deltas();
build_heartbeat(&test_config(), &mut store);
store
.add_volume(
VolumeId(3),
"pics",
None,
None,
0,
DiskType::HardDrive,
Version::current(),
)
.unwrap();
// What the notify path does: collect a snapshot, send a message of its own.
let snapshot =
build_heartbeat_with_ec_status(&test_config(), &mut store, Vec::new(), true, false).1;
assert_eq!(snapshot.len(), 3);
let heartbeat = build_heartbeat(&test_config(), &mut store);
assert_eq!(heartbeat.changed_volumes.len(), 1);
assert_eq!(heartbeat.changed_volumes[0].id, 3);
}
#[test]
fn test_build_heartbeat_digests_exactly_what_it_reports() {
let temp_dir = tempfile::tempdir().unwrap();
let dir = temp_dir.path().to_str().unwrap();
let mut store = Store::new(NeedleMapKind::InMemory);
store
.add_location(
dir,
dir,
8,
DiskType::HardDrive,
MinFreeSpace::Percent(1.0),
Vec::new(),
)
.unwrap();
let empty = build_heartbeat(&test_config(), &mut store);
assert_eq!(empty.volume_digest, Some(0));
for vid in [VolumeId(1), VolumeId(2)] {
store
.add_volume(
vid,
"pics",
None,
None,
0,
DiskType::HardDrive,
Version::current(),
)
.unwrap();
}
let heartbeat = build_heartbeat(&test_config(), &mut store);
assert_eq!(heartbeat.volumes.len(), 2);
let expected = heartbeat
.volumes
.iter()
.fold(0u64, |acc, m| acc ^ crate::storage::volume_report_hash::report_hash(m));
assert_eq!(heartbeat.volume_digest, Some(expected));
assert_ne!(heartbeat.volume_digest, Some(0));
}
#[test]
fn test_build_heartbeat_tracks_go_read_only_labels_and_disk_id() {
let temp_dir = tempfile::tempdir().unwrap();
let dir = temp_dir.path().to_str().unwrap();
let mut store = Store::new(NeedleMapKind::InMemory);
store
.add_location(
dir,
dir,
8,
DiskType::HardDrive,
MinFreeSpace::Percent(1.0),
Vec::new(),
)
.unwrap();
store
.add_volume(
VolumeId(17),
"heartbeat_metrics_case",
None,
None,
0,
DiskType::HardDrive,
Version::current(),
)
.unwrap();
store.locations[0]
.is_disk_space_low
.store(true, Ordering::Relaxed);
{
let (_, volume) = store.find_volume_mut(VolumeId(17)).unwrap();
volume.set_read_only().unwrap();
volume.volume_info.files.push(Default::default());
volume.refresh_remote_write_mode();
}
let heartbeat = build_heartbeat(&test_config(), &mut store);
let collection = "heartbeat_metrics_case";
let disk_type = store.locations[0].disk_type.to_string();
assert_eq!(heartbeat.volumes.len(), 1);
assert_eq!(heartbeat.volumes[0].disk_id, 0);
assert_eq!(heartbeat.max_volume_counts[&disk_type], 1);
assert_eq!(
READ_ONLY_VOLUME_GAUGE
.with_label_values(&[collection, READ_ONLY_LABEL_IS_READ_ONLY])
.get(),
1.0
);
assert_eq!(
READ_ONLY_VOLUME_GAUGE
.with_label_values(&[collection, READ_ONLY_LABEL_NO_WRITE_OR_DELETE])
.get(),
0.0
);
assert_eq!(
READ_ONLY_VOLUME_GAUGE
.with_label_values(&[collection, READ_ONLY_LABEL_NO_WRITE_CAN_DELETE])
.get(),
1.0
);
assert_eq!(
READ_ONLY_VOLUME_GAUGE
.with_label_values(&[collection, READ_ONLY_LABEL_IS_DISK_SPACE_LOW])
.get(),
1.0
);
assert_eq!(
DISK_SIZE_GAUGE
.with_label_values(&[collection, DISK_SIZE_LABEL_NORMAL])
.get(),
crate::storage::super_block::SUPER_BLOCK_SIZE as f64
);
assert_eq!(
DISK_SIZE_GAUGE
.with_label_values(&[collection, DISK_SIZE_LABEL_DELETED_BYTES])
.get(),
0.0
);
}
#[test]
fn test_build_heartbeat_reports_disk_bytes() {
let temp_dir = tempfile::tempdir().unwrap();
let dir = temp_dir.path().to_str().unwrap();
let mut store = Store::new(NeedleMapKind::InMemory);
store
.add_location(
dir,
dir,
8,
DiskType::HardDrive,
MinFreeSpace::Percent(1.0),
Vec::new(),
)
.unwrap();
// Populate the cached physical-capacity fields from a real statvfs probe.
store.locations[0].check_disk_space();
let heartbeat = build_heartbeat(&test_config(), &mut store);
let disk_type = store.locations[0].disk_type.to_string();
assert!(
heartbeat
.disk_total_bytes
.get(&disk_type)
.copied()
.unwrap_or(0)
> 0,
"expected nonzero disk_total_bytes for the temp filesystem"
);
assert!(
heartbeat.disk_free_bytes.contains_key(&disk_type),
"expected a disk_free_bytes entry for the disk type"
);
}
#[test]
fn test_collect_ec_heartbeat_sets_go_metadata_and_ec_metrics() {
let temp_dir = tempfile::tempdir().unwrap();
let dir = temp_dir.path().to_str().unwrap();
let mut store = Store::new(NeedleMapKind::InMemory);
store
.add_location(
dir,
dir,
8,
DiskType::HardDrive,
MinFreeSpace::Percent(1.0),
Vec::new(),
)
.unwrap();
let shard_path = format!("{}/ec_metrics_case_27.ec00", dir);
std::fs::write(&shard_path, b"ec-shard").unwrap();
store.locations[0]
.mount_ec_shards(VolumeId(27), "ec_metrics_case", &[0], "")
.unwrap();
let state = test_state_with_store(store);
let heartbeat = collect_ec_heartbeat(&test_config(), &state);
assert_eq!(heartbeat.ec_shards.len(), 1);
assert!(!heartbeat.has_no_ec_shards);
assert_eq!(heartbeat.ec_shards[0].disk_id, 0);
assert_eq!(
heartbeat.ec_shards[0].disk_type,
state.store.read().unwrap().locations[0].disk_type.to_string()
);
assert_eq!(heartbeat.ec_shards[0].ec_index_bits, 1);
assert_eq!(heartbeat.ec_shards[0].shard_sizes, vec![8]);
assert_eq!(
DISK_SIZE_GAUGE
.with_label_values(&["ec_metrics_case", DISK_SIZE_LABEL_EC])
.get(),
8.0
);
}
#[test]
fn test_collect_heartbeat_deletes_expired_ec_volumes() {
let temp_dir = tempfile::tempdir().unwrap();
let dir = temp_dir.path().to_str().unwrap();
let mut store = Store::new(NeedleMapKind::InMemory);
store
.add_location(
dir,
dir,
8,
DiskType::HardDrive,
MinFreeSpace::Percent(1.0),
Vec::new(),
)
.unwrap();
std::fs::write(format!("{}/expired_heartbeat_ec_31.ec00", dir), b"expired").unwrap();
store.locations[0]
.mount_ec_shards(VolumeId(31), "expired_heartbeat_ec", &[0], "")
.unwrap();
store
.find_ec_volume_mut(VolumeId(31))
.unwrap()
.expire_at_sec = 1;
let state = test_state_with_store(store);
let heartbeat = collect_heartbeat(&test_config(), &state);
assert!(heartbeat.has_no_ec_shards);
assert_eq!(heartbeat.deleted_ec_shards.len(), 1);
assert_eq!(heartbeat.deleted_ec_shards[0].id, 31);
assert!(!state.store.read().unwrap().has_ec_volume(VolumeId(31)));
}
#[test]
fn test_collect_heartbeat_excludes_expired_volume_until_removal_delay() {
let temp_dir = tempfile::tempdir().unwrap();
let dir = temp_dir.path().to_str().unwrap();
let mut store = Store::new(NeedleMapKind::InMemory);
store
.add_location(
dir,
dir,
8,
DiskType::HardDrive,
MinFreeSpace::Percent(1.0),
Vec::new(),
)
.unwrap();
store.volume_size_limit.store(1, Ordering::Relaxed);
store
.add_volume(
VolumeId(41),
"expired_volume_case",
None,
Some(crate::storage::needle::ttl::TTL::read("20m").unwrap()),
1024,
DiskType::HardDrive,
Version::current(),
)
.unwrap();
let dat_path = {
let (_, volume) = store.find_volume_mut(VolumeId(41)).unwrap();
volume.set_last_io_error_for_test(None);
volume.set_last_modified_ts_for_test(
SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap_or_default()
.as_secs()
.saturating_sub(21 * 60),
);
volume.dat_path()
};
std::fs::OpenOptions::new()
.write(true)
.open(&dat_path)
.unwrap()
.set_len((crate::storage::super_block::SUPER_BLOCK_SIZE + 1) as u64)
.unwrap();
let volume_size_limit = store.volume_size_limit.load(Ordering::Relaxed);
let (_, volume) = store.find_volume(VolumeId(41)).unwrap();
assert!(volume.is_expired(volume.dat_file_size().unwrap_or(0), volume_size_limit));
assert!(!volume.is_expired_long_enough(10));
let heartbeat = build_heartbeat(&test_config(), &mut store);
assert!(heartbeat.volumes.is_empty());
assert!(store.has_volume(VolumeId(41)));
}
#[test]
fn test_collect_heartbeat_deletes_io_error_volume() {
let temp_dir = tempfile::tempdir().unwrap();
let dir = temp_dir.path().to_str().unwrap();
let mut store = Store::new(NeedleMapKind::InMemory);
store
.add_location(
dir,
dir,
8,
DiskType::HardDrive,
MinFreeSpace::Percent(1.0),
Vec::new(),
)
.unwrap();
store
.add_volume(
VolumeId(51),
"io_error_case",
None,
None,
0,
DiskType::HardDrive,
Version::current(),
)
.unwrap();
let (_, volume) = store.find_volume_mut(VolumeId(51)).unwrap();
volume.set_last_io_error_for_test(Some("input/output error"));
let heartbeat = build_heartbeat(&test_config(), &mut store);
assert!(heartbeat.volumes.is_empty());
assert!(!store.has_volume(VolumeId(51)));
}
#[test]
fn test_build_heartbeat_includes_remote_storage_name_and_key() {
let temp_dir = tempfile::tempdir().unwrap();
let dir = temp_dir.path().to_str().unwrap();
let mut store = Store::new(NeedleMapKind::InMemory);
store
.add_location(
dir,
dir,
8,
DiskType::HardDrive,
MinFreeSpace::Percent(1.0),
Vec::new(),
)
.unwrap();
store
.add_volume(
VolumeId(71),
"remote_volume_case",
None,
None,
0,
DiskType::HardDrive,
Version::current(),
)
.unwrap();
let (_, volume) = store.find_volume_mut(VolumeId(71)).unwrap();
volume.volume_info.files.push(crate::storage::volume::PbRemoteFile {
backend_type: "s3".to_string(),
backend_id: "archive".to_string(),
key: "volumes/71.dat".to_string(),
..Default::default()
});
volume.refresh_remote_write_mode();
let heartbeat = build_heartbeat(&test_config(), &mut store);
assert_eq!(heartbeat.volumes.len(), 1);
assert_eq!(heartbeat.volumes[0].remote_storage_name, "s3.archive");
assert_eq!(heartbeat.volumes[0].remote_storage_key, "volumes/71.dat");
}
#[test]
fn test_apply_storage_backends_registers_s3_default_aliases() {
let state = test_state_with_store(Store::new(NeedleMapKind::InMemory));
// Do not call clear() on the global registry — other tests may be
// running concurrently. Just register our entries and verify them.
apply_storage_backends(
&state,
&[master_pb::StorageBackend {
r#type: "s3".to_string(),
id: "default".to_string(),
properties: std::collections::HashMap::from([
("aws_access_key_id".to_string(), "access".to_string()),
("aws_secret_access_key".to_string(), "secret".to_string()),
("bucket".to_string(), "bucket-a".to_string()),
("region".to_string(), "us-west-2".to_string()),
("endpoint".to_string(), "http://127.0.0.1:8333".to_string()),
("storage_class".to_string(), "STANDARD".to_string()),
("force_path_style".to_string(), "false".to_string()),
]),
}],
);
let registry = state.s3_tier_registry.read().unwrap();
assert!(registry.get("s3.default").is_some());
assert!(registry.get("s3").is_some());
let global_registry = crate::remote_storage::s3_tier::global_s3_tier_registry()
.read()
.unwrap();
assert!(global_registry.get("s3.default").is_some());
assert!(global_registry.get("s3").is_some());
}
#[test]
fn test_apply_storage_backends_ignores_unsupported_types() {
let state = test_state_with_store(Store::new(NeedleMapKind::InMemory));
// Do not call clear() on the global registry — other tests may be
// running concurrently.
apply_storage_backends(
&state,
&[master_pb::StorageBackend {
r#type: "rclone".to_string(),
id: "default".to_string(),
properties: std::collections::HashMap::new(),
}],
);
// The per-state registry is freshly created and should have no entries
// since "rclone" is unsupported.
let registry = state.s3_tier_registry.read().unwrap();
assert!(registry.names().is_empty());
// Only check that the unsupported type was not added to the global
// registry. Other tests may have their own entries present.
let global_registry = crate::remote_storage::s3_tier::global_s3_tier_registry()
.read()
.unwrap();
assert!(global_registry.get("rclone.default").is_none());
assert!(global_registry.get("rclone").is_none());
}
#[test]
fn test_apply_metrics_push_settings_updates_runtime_state() {
let store = Store::new(NeedleMapKind::InMemory);
let state = test_state_with_store(store);
assert!(apply_metrics_push_settings(&state, "pushgateway:9091", 15,));
{
let runtime = state.metrics_runtime.read().unwrap();
assert_eq!(runtime.push_gateway.address, "pushgateway:9091");
assert_eq!(runtime.push_gateway.interval_seconds, 15);
}
assert!(!apply_metrics_push_settings(&state, "pushgateway:9091", 15,));
}
#[test]
fn test_duplicate_uuid_retry_delay_matches_go_backoff() {
assert_eq!(duplicate_uuid_retry_delay(0), Duration::from_secs(2));
assert_eq!(duplicate_uuid_retry_delay(1), Duration::from_secs(4));
assert_eq!(duplicate_uuid_retry_delay(2), Duration::from_secs(8));
}
#[test]
fn test_duplicate_directories_maps_master_uuids_to_paths() {
let temp_dir = tempfile::tempdir().unwrap();
let dir = temp_dir.path().to_str().unwrap();
let mut store = Store::new(NeedleMapKind::InMemory);
store
.add_location(
dir,
dir,
1,
DiskType::HardDrive,
MinFreeSpace::Percent(1.0),
Vec::new(),
)
.unwrap();
let duplicate_dirs = duplicate_directories(
&store,
&[
store.locations[0].directory_uuid.clone(),
"missing-uuid".to_string(),
],
);
assert_eq!(duplicate_dirs, vec![dir.to_string()]);
}
#[test]
fn test_apply_master_volume_options_updates_preallocate_and_size_limit() {
let store = Store::new(NeedleMapKind::InMemory);
store.volume_size_limit.store(1024, Ordering::Relaxed);
let changed = apply_master_volume_options(
&store,
&master_pb::HeartbeatResponse {
volume_size_limit: 2048,
preallocate: true,
..Default::default()
},
);
assert!(store.get_preallocate());
assert_eq!(store.volume_size_limit.load(Ordering::Relaxed), 2048);
assert!(!changed);
}
#[test]
fn test_diff_ec_shard_delta_messages_reports_mounts_and_unmounts() {
let temp_dir = tempfile::tempdir().unwrap();
let dir = temp_dir.path().to_str().unwrap();
let mut store = Store::new(NeedleMapKind::InMemory);
store
.add_location(
dir,
dir,
8,
DiskType::HardDrive,
MinFreeSpace::Percent(1.0),
Vec::new(),
)
.unwrap();
let previous = collect_ec_shard_delta_messages(&store);
std::fs::write(format!("{}/ec_delta_case_81.ec00", dir), b"delta").unwrap();
store.locations[0]
.mount_ec_shards(VolumeId(81), "ec_delta_case", &[0], "")
.unwrap();
let current = collect_ec_shard_delta_messages(&store);
let (new_ec_shards, deleted_ec_shards) =
diff_ec_shard_delta_messages(&previous, &current);
assert_eq!(new_ec_shards.len(), 1);
assert!(deleted_ec_shards.is_empty());
assert_eq!(new_ec_shards[0].ec_index_bits, 1);
assert_eq!(new_ec_shards[0].shard_sizes, vec![5]);
let (new_after_delete, deleted_after_delete) =
diff_ec_shard_delta_messages(&current, &HashMap::new());
assert!(new_after_delete.is_empty());
assert_eq!(deleted_after_delete.len(), 1);
assert_eq!(deleted_after_delete[0].ec_index_bits, 1);
assert_eq!(deleted_after_delete[0].shard_sizes, vec![0]);
}
}
Reference in new issue View Git Blame Copy Permalink