diff --git a/seaweed-volume/src/server/grpc_server.rs b/seaweed-volume/src/server/grpc_server.rs index 4c4918fde..04c051e75 100644 --- a/seaweed-volume/src/server/grpc_server.rs +++ b/seaweed-volume/src/server/grpc_server.rs @@ -1125,7 +1125,9 @@ impl VolumeServer for VolumeGrpcService { let store = self.state.store.read().unwrap(); let garbage_ratio = match store.find_volume(vid) { Some((_, vol)) => vol.garbage_level(), - None => return Err(Status::not_found(format!("not found volume id {}", vid))), + None => { + return Err(crate::storage::volume::VolumeError::VolumeNotFound(vid).into()); + } }; Ok(Response::new(volume_server_pb::VacuumVolumeCheckResponse { garbage_ratio, @@ -1183,7 +1185,10 @@ impl VolumeServer for VolumeGrpcService { .inc(); if let Err(e) = result { - let _ = tx.blocking_send(Err(Status::internal(e))); + let _ = tx.blocking_send(Err(crate::server::status_with_context( + &format!("compact volume {vid}"), + e, + ))); } }); @@ -1225,7 +1230,10 @@ impl VolumeServer for VolumeGrpcService { volume_size, }, )), - Err(e) => Err(Status::internal(e)), + Err(e) => Err(crate::server::status_with_context( + &format!("commit compact volume {vid}"), + e, + )), } } @@ -1241,7 +1249,10 @@ impl VolumeServer for VolumeGrpcService { Ok(()) => Ok(Response::new( volume_server_pb::VacuumVolumeCleanupResponse {}, )), - Err(e) => Err(Status::internal(e)), + Err(e) => Err(crate::server::status_with_context( + &format!("cleanup volume {vid}"), + e, + )), } } @@ -1253,9 +1264,9 @@ impl VolumeServer for VolumeGrpcService { let collection = &request.into_inner().collection; { let mut store = self.state.store.write().unwrap(); - store - .delete_collection(collection) - .map_err(Status::internal)?; + store.delete_collection(collection).map_err(|e| { + crate::server::status_with_context(&format!("delete collection {collection}"), e) + })?; } // The delta the notify path derives is the only thing that tells the // master these slots came free: a heartbeat carries the whole list only @@ -1560,24 +1571,16 @@ impl VolumeServer for VolumeGrpcService { let vid = VolumeId(req.volume_id); let mut store = self.state.store.write().unwrap(); if req.only_empty { - let (_, vol) = store - .find_volume(vid) - .ok_or_else(|| Status::not_found(format!("not found volume id {}", vid)))?; + let (_, vol) = store.find_volume(vid).ok_or_else(|| { + Status::from(crate::storage::volume::VolumeError::VolumeNotFound(vid)) + })?; if vol.file_count() > 0 { - return Err(Status::failed_precondition("volume not empty")); + return Err(Status::from(crate::storage::volume::VolumeError::NotEmpty)); } } store .delete_volume(vid, req.only_empty, req.keep_remote_data) - .map_err(|e| match e { - crate::storage::volume::VolumeError::NotFound => { - Status::not_found(format!("not found volume id {}", vid)) - } - crate::storage::volume::VolumeError::NotEmpty => { - Status::failed_precondition("volume not empty") - } - other => Status::internal(other.to_string()), - })?; + .map_err(|e| crate::server::status_with_context(&format!("delete volume {vid}"), e))?; self.state.volume_state_notify.notify_one(); Ok(Response::new(volume_server_pb::VolumeDeleteResponse {})) } @@ -6575,6 +6578,30 @@ mod tests { assert_eq!(v.file_count(), 1); } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn test_vacuum_volume_commit_missing_volume_is_not_found() { + let (service, _tmp) = make_local_service_with_volume("vacuum_commit_missing", None); + + let mut request = + Request::new(volume_server_pb::VacuumVolumeCommitRequest { volume_id: 4242 }); + request + .extensions_mut() + .insert(tonic::transport::server::TcpConnectInfo { + local_addr: None, + remote_addr: Some("127.0.0.1:65000".parse().unwrap()), + }); + + let err = service + .vacuum_volume_commit(request) + .await + .expect_err("committing a compaction for a volume that is not mounted must fail"); + assert_eq!(err.code(), tonic::Code::NotFound, "{err:?}"); + assert!( + err.message().contains("4242"), + "the message must still name the volume: {err:?}" + ); + } + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] async fn test_volume_incremental_copy_streams_remote_only_volume_data() { let (service, _tmp, shutdown_tx, dat_bytes, super_block_size, _delete_count) = diff --git a/seaweed-volume/src/server/mod.rs b/seaweed-volume/src/server/mod.rs index 092e43c35..16877f064 100644 --- a/seaweed-volume/src/server/mod.rs +++ b/seaweed-volume/src/server/mod.rs @@ -1,3 +1,7 @@ +use tonic::Status; + +use crate::storage::volume::VolumeError; + #[cfg(unix)] pub mod debug; pub mod grpc_client; @@ -13,3 +17,66 @@ pub mod store_ec; pub mod ui; pub mod volume_server; pub mod write_queue; + +/// Map a storage error onto the gRPC code that describes it. +impl From for Status { + fn from(err: VolumeError) -> Self { + let message = err.to_string(); + match err { + VolumeError::NotFound | VolumeError::VolumeNotFound(_) => Status::not_found(message), + VolumeError::ReadOnly | VolumeError::NotEmpty => Status::failed_precondition(message), + VolumeError::InsufficientSpace { .. } => Status::resource_exhausted(message), + VolumeError::AlreadyExists => Status::already_exists(message), + _ => Status::internal(message), + } + } +} + +/// Same mapping, with the RPC's own context prefixed (`compact volume 7: ...`). +pub fn status_with_context(context: &str, err: VolumeError) -> Status { + let status = Status::from(err); + Status::new(status.code(), format!("{context}: {}", status.message())) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::storage::types::VolumeId; + + #[test] + fn test_volume_error_maps_to_grpc_code() { + use tonic::Code; + + let code = |e: VolumeError| Status::from(e).code(); + assert_eq!( + code(VolumeError::VolumeNotFound(VolumeId(7))), + Code::NotFound + ); + assert_eq!(code(VolumeError::NotFound), Code::NotFound); + assert_eq!(code(VolumeError::ReadOnly), Code::FailedPrecondition); + assert_eq!( + code(VolumeError::InsufficientSpace { + vid: VolumeId(7), + required: 2, + free: 1, + }), + Code::ResourceExhausted + ); + assert_eq!(code(VolumeError::AlreadyExists), Code::AlreadyExists); + assert_eq!(code(VolumeError::NotInitialized), Code::Internal); + + let status = status_with_context( + "compact volume 7", + VolumeError::InsufficientSpace { + vid: VolumeId(7), + required: 2, + free: 1, + }, + ); + assert_eq!(status.code(), Code::ResourceExhausted); + assert_eq!( + status.message(), + "compact volume 7: not enough free space: required 2, free 1" + ); + } +} diff --git a/seaweed-volume/src/storage/needle_map.rs b/seaweed-volume/src/storage/needle_map.rs index af25dd59a..a7b7b3592 100644 --- a/seaweed-volume/src/storage/needle_map.rs +++ b/seaweed-volume/src/storage/needle_map.rs @@ -417,9 +417,9 @@ impl CompactNeedleMap { } /// Visit all entries in ascending order by needle ID. - pub fn ascending_visit(&self, f: F) -> Result<(), String> + pub fn ascending_visit(&self, f: F) -> Result<(), E> where - F: FnMut(NeedleId, &NeedleValue) -> Result<(), String>, + F: FnMut(NeedleId, &NeedleValue) -> Result<(), E>, { self.map.ascending_visit(f) } @@ -1203,9 +1203,10 @@ impl RedbNeedleMap { } /// Visit all entries in ascending order by needle ID. - pub fn ascending_visit(&self, mut f: F) -> Result<(), String> + pub fn ascending_visit(&self, mut f: F) -> Result<(), E> where - F: FnMut(NeedleId, &NeedleValue) -> Result<(), String>, + F: FnMut(NeedleId, &NeedleValue) -> Result<(), E>, + E: From, { let txn = self .db_or_err() @@ -1443,9 +1444,10 @@ impl NeedleMap { } /// Visit all entries in ascending order by needle ID. - pub fn ascending_visit(&self, f: F) -> Result<(), String> + pub fn ascending_visit(&self, f: F) -> Result<(), E> where - F: FnMut(NeedleId, &NeedleValue) -> Result<(), String>, + F: FnMut(NeedleId, &NeedleValue) -> Result<(), E>, + E: From, { match self { NeedleMap::InMemory(nm) => nm.ascending_visit(f), @@ -1466,7 +1468,7 @@ impl NeedleMap { // The visitor never fails, so neither can this. let _ = nm.ascending_visit(|id, nv| { entries.push((id, *nv)); - Ok(()) + Ok::<(), std::convert::Infallible>(()) }); Ok(entries) } @@ -1838,7 +1840,7 @@ mod tests { let mut live = 0u64; nm.ascending_visit(|_, _| { live += 1; - Ok(()) + Ok::<(), String>(()) }) .unwrap(); assert_eq!(live, N - 1); @@ -2051,7 +2053,7 @@ mod tests { let mut visited = Vec::new(); nm.ascending_visit(|id, nv| { visited.push((id, nv.size)); - Ok(()) + Ok::<(), String>(()) }) .unwrap(); diff --git a/seaweed-volume/src/storage/needle_map/sorted_file.rs b/seaweed-volume/src/storage/needle_map/sorted_file.rs index 5b2695bd4..ec3f9af0a 100644 --- a/seaweed-volume/src/storage/needle_map/sorted_file.rs +++ b/seaweed-volume/src/storage/needle_map/sorted_file.rs @@ -317,9 +317,11 @@ impl SortedFileNeedleMap { Ok(()) } - pub fn ascending_visit(&self, mut f: F) -> Result<(), String> + /// Visit all live entries in ascending order by needle ID. + pub fn ascending_visit(&self, mut f: F) -> Result<(), E> where - F: FnMut(NeedleId, &NeedleValue) -> Result<(), String>, + F: FnMut(NeedleId, &NeedleValue) -> Result<(), E>, + E: From, { let mut visit_error = None; self.visit_live_entries(|id, nv| { @@ -329,7 +331,7 @@ impl SortedFileNeedleMap { } Ok(()) }) - .map_err(|e| visit_error.take().unwrap_or_else(|| e.to_string())) + .map_err(|e| visit_error.take().unwrap_or_else(|| E::from(e.to_string()))) } pub fn iter_entries(&self) -> io::Result> { @@ -1035,7 +1037,7 @@ mod tests { let mut visited = Vec::new(); m.ascending_visit(|id, _| { visited.push(id); - Ok(()) + Ok::<(), String>(()) }) .unwrap(); assert_eq!(visited, vec![NeedleId(2)]); diff --git a/seaweed-volume/src/storage/store.rs b/seaweed-volume/src/storage/store.rs index c415ac214..df2fa3403 100644 --- a/seaweed-volume/src/storage/store.rs +++ b/seaweed-volume/src/storage/store.rs @@ -764,10 +764,9 @@ impl Store { // ---- Collection operations ---- /// Delete all volumes in a collection. - pub fn delete_collection(&mut self, collection: &str) -> Result<(), String> { + pub fn delete_collection(&mut self, collection: &str) -> Result<(), VolumeError> { for loc in &mut self.locations { - loc.delete_collection(collection) - .map_err(|e| format!("delete collection {}: {}", collection, e))?; + loc.delete_collection(collection)?; } crate::metrics::delete_collection_metrics(collection); Ok(()) @@ -1365,18 +1364,6 @@ impl Store { // ---- Vacuum / Compaction ---- - /// Check the garbage level of a volume. - pub fn check_compact_volume(&self, vid: VolumeId) -> Result { - if let Some((_, v)) = self.find_volume(vid) { - Ok(v.garbage_level()) - } else { - Err(format!( - "volume id {} is not found during check compact", - vid.0 - )) - } - } - /// Compact a volume by rewriting only live needles. pub fn compact_volume( &mut self, @@ -1384,68 +1371,53 @@ impl Store { preallocate: u64, max_bytes_per_second: i64, progress_fn: F, - ) -> Result<(), String> + ) -> Result<(), VolumeError> where F: Fn(i64) -> bool, { - let loc_idx = self - .find_volume(vid) - .map(|(i, _)| i) - .ok_or_else(|| format!("volume id {} is not found during compact", vid.0))?; + // Required space matches Go's CompactVolume check: the larger of the + // requested preallocation and the estimated volume size. + let (loc_idx, space_needed) = { + let (loc_idx, v) = self + .find_volume(vid) + .ok_or(VolumeError::VolumeNotFound(vid))?; + let estimated = v.dat_file_size().unwrap_or(0) + v.idx_file_size(); + (loc_idx, std::cmp::max(preallocate, estimated)) + }; let dir = self.locations[loc_idx].directory.clone(); let (_, free) = crate::storage::disk_location::get_disk_stats(&dir); - - // Compute required space: use the larger of preallocate or estimated volume size - // matching Go's CompactVolume space check - let space_needed = { - let (_, v) = self.find_volume(vid).unwrap(); - let estimated = v.dat_file_size().unwrap_or(0) + v.idx_file_size(); - std::cmp::max(preallocate, estimated) - }; - if free < space_needed { - return Err(format!( - "not enough free space to compact volume {}. Required: {}, Free: {}", - vid.0, space_needed, free - )); + return Err(VolumeError::InsufficientSpace { + vid, + required: space_needed, + free, + }); } - if let Some((_, v)) = self.find_volume_mut(vid) { - v.compact_by_index(preallocate, max_bytes_per_second, progress_fn) - .map_err(|e| format!("compact volume {}: {}", vid.0, e)) - } else { - Err(format!("volume id {} is not found during compact", vid.0)) - } + let (_, v) = self + .find_volume_mut(vid) + .ok_or(VolumeError::VolumeNotFound(vid))?; + v.compact_by_index(preallocate, max_bytes_per_second, progress_fn) } /// Commit a completed compaction: swap files and reload. - pub fn commit_compact_volume(&mut self, vid: VolumeId) -> Result<(bool, u64), String> { - if let Some((_, v)) = self.find_volume_mut(vid) { - let is_read_only = v.is_read_only(); - v.commit_compact() - .map_err(|e| format!("commit compact volume {}: {}", vid.0, e))?; - let volume_size = v.dat_file_size().unwrap_or(0); - Ok((is_read_only, volume_size)) - } else { - Err(format!( - "volume id {} is not found during commit compact", - vid.0 - )) - } + pub fn commit_compact_volume(&mut self, vid: VolumeId) -> Result<(bool, u64), VolumeError> { + let (_, v) = self + .find_volume_mut(vid) + .ok_or(VolumeError::VolumeNotFound(vid))?; + let is_read_only = v.is_read_only(); + v.commit_compact()?; + let volume_size = v.dat_file_size().unwrap_or(0); + Ok((is_read_only, volume_size)) } /// Clean up leftover compaction files. - pub fn cleanup_compact_volume(&mut self, vid: VolumeId) -> Result<(), String> { - if let Some((_, v)) = self.find_volume_mut(vid) { - v.cleanup_compact() - .map_err(|e| format!("cleanup volume {}: {}", vid.0, e)) - } else { - Err(format!( - "volume id {} is not found during cleaning up", - vid.0 - )) - } + pub fn cleanup_compact_volume(&mut self, vid: VolumeId) -> Result<(), VolumeError> { + let (_, v) = self + .find_volume_mut(vid) + .ok_or(VolumeError::VolumeNotFound(vid))?; + v.cleanup_compact() } /// Close all locations and their volumes. @@ -2324,6 +2296,88 @@ mod tests { assert!(matches!(err, Err(VolumeError::NotFound))); } + #[test] + fn test_compaction_of_missing_volume_is_volume_not_found() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let mut store = make_test_store(&[dir]); + + let missing = VolumeId(4242); + + let compact = store.compact_volume(missing, 0, 0, |_| true); + assert!( + matches!(compact, Err(VolumeError::VolumeNotFound(v)) if v == missing), + "{compact:?}" + ); + + let commit = store.commit_compact_volume(missing); + assert!( + matches!(commit, Err(VolumeError::VolumeNotFound(v)) if v == missing), + "{commit:?}" + ); + + let cleanup = store.cleanup_compact_volume(missing); + assert!( + matches!(cleanup, Err(VolumeError::VolumeNotFound(v)) if v == missing), + "{cleanup:?}" + ); + + assert_eq!( + VolumeError::VolumeNotFound(missing).to_string(), + "volume id 4242 is not found" + ); + } + + #[test] + fn test_compact_then_commit_reclaims_a_deleted_needle() { + let tmp = TempDir::new().unwrap(); + let dir = tmp.path().to_str().unwrap(); + let mut store = make_test_store(&[dir]); + + let vid = VolumeId(1); + store + .add_volume(vid, DiskType::HardDrive, &VolumeSpec::default()) + .unwrap(); + + for i in 1..=3u64 { + let payload = format!("data-{i}").into_bytes(); + let mut n = Needle { + id: NeedleId(i), + cookie: Cookie(i as u32), + data_size: payload.len() as u32, + data: payload, + ..Needle::default() + }; + store.write_volume_needle(vid, &mut n, true).unwrap(); + } + + let mut del = Needle { + id: NeedleId(2), + cookie: Cookie(2), + ..Needle::default() + }; + store.delete_volume_needle(vid, &mut del).unwrap(); + + let size_before = store.find_volume(vid).unwrap().1.dat_file_size().unwrap(); + + // preallocate 0, unthrottled, progress fn that never cancels. + store.compact_volume(vid, 0, 0, |_| true).unwrap(); + + let (is_read_only, volume_size) = store.commit_compact_volume(vid).unwrap(); + assert!(!is_read_only); + assert!( + volume_size < size_before, + "compaction must drop the deleted needle: {volume_size} vs {size_before}" + ); + + let (_, v) = store + .find_volume(vid) + .expect("the volume stays mounted across a commit"); + assert_eq!(v.file_count(), 2); + assert_eq!(v.deleted_count(), 0); + assert_eq!(v.dat_file_size().unwrap(), volume_size); + } + /// Build a Store with N HDD disk locations under a single TempDir. /// Returns the store and the TempDir guard so callers keep the dirs /// alive for the test's lifetime. diff --git a/seaweed-volume/src/storage/volume.rs b/seaweed-volume/src/storage/volume.rs index c12181833..7e0f811a5 100644 --- a/seaweed-volume/src/storage/volume.rs +++ b/seaweed-volume/src/storage/volume.rs @@ -38,6 +38,9 @@ pub enum VolumeError { #[error("not found")] NotFound, + #[error("volume id {0} is not found")] + VolumeNotFound(VolumeId), + #[error("already deleted")] Deleted, @@ -62,6 +65,13 @@ pub enum VolumeError { #[error("volume size limit exceeded: current {current}, limit {limit}")] SizeLimitExceeded { current: u64, limit: u64 }, + #[error("not enough free space: required {required}, free {free}")] + InsufficientSpace { + vid: VolumeId, + required: u64, + free: u64, + }, + #[error("volume not initialized")] NotInitialized, diff --git a/weed/server/volume_grpc_admin.go b/weed/server/volume_grpc_admin.go index 4699fbfd2..f2b1eb286 100644 --- a/weed/server/volume_grpc_admin.go +++ b/weed/server/volume_grpc_admin.go @@ -82,9 +82,9 @@ func (vs *VolumeServer) DeleteCollection(ctx context.Context, req *volume_server if err != nil { glog.Errorf("delete collection %s: %v", req.Collection, err) - } else { - glog.V(2).Infof("delete collection %v", req) + return resp, volumeStatusError(fmt.Errorf("delete collection %s: %w", req.Collection, err)) } + glog.V(2).Infof("delete collection %v", req) return resp, err @@ -202,7 +202,7 @@ func (vs *VolumeServer) VolumeDelete(ctx context.Context, req *volume_server_pb. if err != nil { glog.Errorf("volume delete %v: %v", req, err) - return resp, volumeDeleteStatusError(err) + return resp, volumeStatusError(err) } else { // V(0) so destructive RPCs are always traceable. glog.Infof("volume delete %v", req) @@ -212,16 +212,19 @@ func (vs *VolumeServer) VolumeDelete(ctx context.Context, req *volume_server_pb. } -// volumeDeleteStatusError keeps the store's message so callers matching on +// volumeStatusError keeps the store's message so callers matching on // "not found" or "volume not empty" keep working, and adds the status code so // new callers do not have to. -func volumeDeleteStatusError(err error) error { +func volumeStatusError(err error) error { if errors.Is(err, storage.ErrVolumeNotFound) { return status.Error(codes.NotFound, err.Error()) } if errors.Is(err, storage.ErrVolumeNotEmpty) { return status.Error(codes.FailedPrecondition, err.Error()) } + if errors.Is(err, storage.ErrInsufficientSpace) { + return status.Error(codes.ResourceExhausted, err.Error()) + } return err } diff --git a/weed/server/volume_grpc_admin_delete_test.go b/weed/server/volume_grpc_admin_delete_test.go index 721340f13..72c35fdb6 100644 --- a/weed/server/volume_grpc_admin_delete_test.go +++ b/weed/server/volume_grpc_admin_delete_test.go @@ -15,18 +15,22 @@ import ( "github.com/stretchr/testify/require" ) -func TestVolumeDeleteStatusErrorDistinguishesAbsentFromTransportFailure(t *testing.T) { - notFound := volumeDeleteStatusError(fmt.Errorf("delete volume 17 not found on disk: %w", storage.ErrVolumeNotFound)) +func TestVolumeStatusErrorDistinguishesAbsentFromTransportFailure(t *testing.T) { + notFound := volumeStatusError(fmt.Errorf("delete volume 17 not found on disk: %w", storage.ErrVolumeNotFound)) assert.Equal(t, codes.NotFound, status.Code(notFound)) assert.Contains(t, notFound.Error(), "not found", "the store message must survive for callers that match on it") transport := errors.New("connection reset") - require.ErrorIs(t, volumeDeleteStatusError(transport), transport) + require.ErrorIs(t, volumeStatusError(transport), transport) assert.NotEqual(t, codes.NotFound, status.Code(transport)) - notEmpty := volumeDeleteStatusError(storage.ErrVolumeNotEmpty) + notEmpty := volumeStatusError(storage.ErrVolumeNotEmpty) assert.Equal(t, codes.FailedPrecondition, status.Code(notEmpty)) assert.Contains(t, notEmpty.Error(), "volume not empty") + + noSpace := volumeStatusError(fmt.Errorf("compact volume 17: %w", storage.ErrInsufficientSpace)) + assert.Equal(t, codes.ResourceExhausted, status.Code(noSpace)) + assert.Contains(t, noSpace.Error(), "insufficient free space") } func TestVolumeDeleteMapsAbsentStoreVolumeToNotFound(t *testing.T) { diff --git a/weed/server/volume_grpc_vacuum.go b/weed/server/volume_grpc_vacuum.go index 9ee88d6f8..7bea72497 100644 --- a/weed/server/volume_grpc_vacuum.go +++ b/weed/server/volume_grpc_vacuum.go @@ -2,6 +2,7 @@ package weed_server import ( "context" + "fmt" "strconv" "time" @@ -27,9 +28,10 @@ func (vs *VolumeServer) VacuumVolumeCheck(ctx context.Context, req *volume_serve if err != nil { glog.V(3).Infof("check volume %d: %v", req.VolumeId, err) + return resp, volumeStatusError(err) } - return resp, err + return resp, nil } @@ -70,7 +72,7 @@ func (vs *VolumeServer) VacuumVolumeCompact(req *volume_server_pb.VacuumVolumeCo stats.VolumeServerVacuumingCompactCounter.WithLabelValues(strconv.FormatBool(err == nil && sendErr == nil)).Inc() if err != nil { glog.Errorf("failed compact volume %d: %v", req.VolumeId, err) - return err + return volumeStatusError(fmt.Errorf("compact volume %d: %w", req.VolumeId, err)) } if sendErr != nil { glog.Errorf("failed compact volume %d report progress: %v", req.VolumeId, sendErr) @@ -99,15 +101,15 @@ func (vs *VolumeServer) VacuumVolumeCommit(ctx context.Context, req *volume_serv readOnly, volumeSize, err := vs.store.CommitCompactVolume(needle.VolumeId(req.VolumeId)) - if err != nil { - glog.Errorf("failed commit volume %d: %v", req.VolumeId, err) - } else { - glog.V(1).Infof("commit volume %d", req.VolumeId) - } stats.VolumeServerVacuumingCommitCounter.WithLabelValues(strconv.FormatBool(err == nil)).Inc() resp.IsReadOnly = readOnly resp.VolumeSize = uint64(volumeSize) - return resp, err + if err != nil { + glog.Errorf("failed commit volume %d: %v", req.VolumeId, err) + return resp, volumeStatusError(fmt.Errorf("commit compact volume %d: %w", req.VolumeId, err)) + } + glog.V(1).Infof("commit volume %d", req.VolumeId) + return resp, nil } @@ -124,10 +126,10 @@ func (vs *VolumeServer) VacuumVolumeCleanup(ctx context.Context, req *volume_ser if err != nil { glog.Errorf("failed cleanup volume %d: %v", req.VolumeId, err) - } else { - glog.V(1).Infof("cleanup volume %d", req.VolumeId) + return resp, volumeStatusError(fmt.Errorf("cleanup volume %d: %w", req.VolumeId, err)) } + glog.V(1).Infof("cleanup volume %d", req.VolumeId) - return resp, err + return resp, nil } diff --git a/weed/storage/store_vacuum.go b/weed/storage/store_vacuum.go index 17e38a989..8eb69df62 100644 --- a/weed/storage/store_vacuum.go +++ b/weed/storage/store_vacuum.go @@ -9,12 +9,14 @@ import ( "github.com/seaweedfs/seaweedfs/weed/storage/needle" ) +var ErrInsufficientSpace = fmt.Errorf("insufficient free space") + func (s *Store) CheckCompactVolume(volumeId needle.VolumeId) (float64, error) { if v := s.findVolume(volumeId); v != nil { glog.V(3).Infof("volume %d garbage level: %f", volumeId, v.garbageLevel()) return v.garbageLevel(), nil } - return 0, fmt.Errorf("volume id %d is not found during check compact", volumeId) + return 0, fmt.Errorf("volume id %d is not found during check compact: %w", volumeId, ErrVolumeNotFound) } func (s *Store) CompactVolume(vid needle.VolumeId, preallocate int64, compactionBytePerSecond int64, progressFn ProgressFunc) error { @@ -28,7 +30,7 @@ func (s *Store) CompactVolume(vid needle.VolumeId, preallocate int64, compaction ProgressCallback: progressFn, }) } - return fmt.Errorf("volume id %d is not found during compact", vid) + return fmt.Errorf("volume id %d is not found during compact: %w", vid, ErrVolumeNotFound) } func (s *Store) CommitCompactVolume(vid needle.VolumeId) (bool, int64, error) { @@ -44,14 +46,14 @@ func (s *Store) CommitCompactVolume(vid needle.VolumeId) (bool, int64, error) { } return isReadOnly, volumeSize, err } - return false, 0, fmt.Errorf("volume id %d is not found during commit compact", vid) + return false, 0, fmt.Errorf("volume id %d is not found during commit compact: %w", vid, ErrVolumeNotFound) } func (s *Store) CommitCleanupVolume(vid needle.VolumeId) error { if v := s.findVolume(vid); v != nil { return v.cleanupCompact() } - return fmt.Errorf("volume id %d is not found during cleaning up", vid) + return fmt.Errorf("volume id %d is not found during cleaning up: %w", vid, ErrVolumeNotFound) } func ensureCompactVolumeSpace(v *Volume, preallocate int64) error { @@ -69,8 +71,8 @@ func ensureCompactVolumeSpace(v *Volume, preallocate int64) error { diskStatus := stats.NewDiskStatus(v.dir) if int64(diskStatus.Free) < spaceNeeded { - return fmt.Errorf("insufficient free space for compaction: need %d bytes (volume: %d, index: %d), but only %d bytes available", - spaceNeeded, volumeSize, indexSize, diskStatus.Free) + return fmt.Errorf("insufficient free space for compaction: need %d bytes (volume: %d, index: %d), but only %d bytes available: %w", + spaceNeeded, volumeSize, indexSize, diskStatus.Free, ErrInsufficientSpace) } glog.V(1).Infof("volume %d compaction space check: volume=%d, index=%d, space_needed=%d, free_space=%d",