mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-10-27 15:30:45 +01:00
* fix(volume): keep preallocated growth above -volume.minFreeSpace When free space fell below -volume.minFreeSpace, every volume went read-only and the master kept growing new volumes. With preallocation on, each new volume reserved its full size, driving the disk further below the floor (15 Oct incident: 66 empty volumes reserved about 1980 GiB). Two causes, both in weed/storage/store.go: - MaybeAdjustVolumeMax sized a disk's volume slots from raw free space, so space inside the floor was counted as available. It now counts only free space above the floor. - addVolume trusted the once-a-minute low-disk flag. A preallocated volume is now refused when the disk is low now, or when its reservation would not fit above the floor. Volumes without preallocation, including merges that copy existing data, are not affected. addVolume runs under a store-level lock, so two concurrent creates cannot both pass the room check against the same free space. * volume: mirror headroom guard in the rust store find_free_location only trusted the once-a-probe is_disk_space_low flag, so a burst of preallocated creates could still run the disk under -volume.minFreeSpace between probes, and maybe_adjust_volume_max sized slots from raw free space. Adds MinFreeSpace::is_low/available_space to match Go and applies both checks the same way. store.write() already serializes add_volume, so no extra lock is needed. * volume: read rust headroom tests from a fake disk The floor arithmetic used a live free-space reading, so unrelated disk activity could flip a test between pass and fail. Store.disk_stats_of is now the same seam Go uses (diskStatusOf): the growth-loop and floor tests run against a controlled disk, mirroring the Go suite one for one. * volume: hold back create metadata from the headroom check An exact-fit preallocation left free space exactly at the floor, so the .vif and .idx writes that follow a create could dip under it and get the location flagged read-only at the next probe. Reserve room for .vif on the data filesystem, and for .idx growth only when dir_idx shares it. * volume: mirror the metadata allowance in the rust store Same rule as the Go side: charge .vif against the checked filesystem and .idx only when dir_idx shares it, and grow the exact-fit test's fake disk by the per-create cost. * volume: skip the reservation check where create cannot preallocate * volume: mirror the platform preallocation gate in the rust store * volume.copy: count the preallocation reservation in the destination space check The copy path picked a location with room for the copied .dat, then reserved the master's full volume size limit on it. A volume smaller than the limit could land on a disk whose free space above -volume.minFreeSpace fit the copy but not the reservation, driving the disk under the floor the same way the growth loop did. Fetch the preallocation setting before choosing the location and require room for the reservation itself, only where CreateVolumeFile actually reserves blocks. * seaweed-volume: store min-free-space-adjusted available space like Go Go's DiskLocation.AvailableSpace holds free space above the -volume.minFreeSpace floor; the Rust probe cached raw free instead, so the VolumeCopy location check could accept a disk already inside the floor. Store the same adjusted figure and cover it with a test. * seaweed-volume: reserve the copied .dat like Go's CreateVolumeFile VolumeCopy called set_len for the reservation, which only grows a sparse file — no blocks are held, so a disk with room for the copied bytes but not the full reservation only fails mid-copy. preallocate_file fallocates KEEP_SIZE on Linux like Go, reserving the blocks up front; off Linux it is a no-op, matching PREALLOCATE_SUPPORTED. * seaweed-volume: charge a copy's destination pick for the .dat reservation Go's VolumeCopy now sizes its free-location check against the master's volume size limit when the platform preallocates, not just the copied bytes. The Rust pick only measured the copied bytes, so a destination could accept a copy whose reservation then dips under the floor. The config fetch moves ahead of the pick (same order as Go) and feeds both the space check and the reservation itself. * volume.copy: charge the picked location's free space under volumeCreateLock The copy path picked a destination off the once-a-minute AvailableSpace probe and released the decision long before the .dat reservation landed. Two copies could pass the same cached headroom and together drive the disk under -volume.minFreeSpace. ReserveFreeLocation serializes the pick under volumeCreateLock, verifies the disk still has live room for the reservation, and charges it against the location's cached free space until the next probe refreshes it. * seaweed-volume: mirror the copy pick's live-check and headroom charge Go's VolumeCopy now picks through Store.ReserveFreeLocation: live disk stats under the store write lock, then the reservation charged against the location's cached free space. The Rust plan_copy_destination gains the same store.reserve_free_location so two copies cannot spend the same headroom before the first .dat reservation lands. * volume.copy: saturate the free-space charge against a live-space drop The two statfs reads bracketing a pick can race a large external write; a plain subtraction would underflow and leave the location looking roomy until the next probe. * volume.copy: release the headroom reservation when a copy does not land ReserveFreeLocation charges the picked location's cached free space, but a failed copy never returned it, so the headroom stayed booked until the next disk-space probe and blocked retries. ReleaseFreeLocation refunds the charge; the copy's cleanup defer now registers with the reservation so every failure path (existing-replica delete, .note write, stream send) releases it, including the post-transfer mount/validate failures covered by the outer defer. * seaweed-volume: mirror the copy headroom release on failed copies release_free_location refunds reserve_free_location's charge. The copy returns it when the plan's existing-replica delete fails, when the .note write fails, and in cleanup_failed_copy, so a failed copy stops holding the headroom until the next disk-space probe. * volume.copy: keep pending copy headroom out of probe readings ReleaseFreeLocation refunded the reservation into AvailableSpace even when a CheckDiskSpace probe had already rebased the cache on live stats — the probe's reading still shows the in-flight copy's bytes as free, so the refund double-counted headroom and two copies could spend the same space, driving the disk under -volume.minFreeSpace. Track the outstanding charge in DiskLocation.ReservedSpace instead: probes subtract it so they cannot re-spend it, a failed copy removes only its own charge, and a landed copy commits it — the charge and the new usage cancel out. hasRoomForNewVolume respects pending headroom too, so a preallocating create cannot overrun an in-flight copy. * seaweed-volume: mirror pending copy headroom across disk probes reserved_space tracks the in-flight copy's charge separately from the cached free space: check_disk_space subtracts it, release removes only the failed copy's charge, a landed copy commits it, and has_room_for_new_volume respects it — matching the Go side. * volume.copy: settle a copy's headroom as its bytes land on disk The pending charge held the copy's full reservation until commit, but written and fallocated bytes already show up in statfs, so the live-space check counted them twice. A slow copy could then block other copies and volume growth despite free space above -volume.minFreeSpace. Settle the charge instead: commit the bytes CreateVolumeFile actually reserved (a failed fallocate leaves nothing committed), then settle each file's write progress as it lands. The failure paths now release only the unsettled remainder; allocatedSize is exported for the .dat's st_blocks. * seaweed-volume: mirror copy headroom settling as bytes land reserved_space keeps only the promised bytes statfs cannot see yet: preallocate_dat settles whatever blocks the filesystem actually reserved (0 when fallocate cannot), each copied file's progress settles the rest as it lands, and a failed copy releases only the unsettled remainder — matching the Go side. * volume.copy: release a failed copy's headroom charge only once Both defers fired on the same in-closure error: the inner released the unsettled charge and the outer released it again, driving ReservedSpace negative (an atomic.Uint64 underflow) and inflating AvailableSpace — the location would then fail every later headroom check, or worse, pass them with fabricated space. Zero the charge once it is released; the Rust mirror already releases via swap(0) for the same reason. * volume.copy: write into the preallocated .dat instead of staging over it The .dat stream staged into .copying and renamed over the file CreateVolumeFile had just fallocated, so the copy's transient footprint was preallocate+dat — more than the reservation the pick just checked — and the reservation itself was discarded at rename. Writing in place reuses the fallocated extent, keeps the peak at the reserved maximum, and leaves the copied volume preallocated like a created one. * seaweed-volume: mirror the in-place write into the preallocated .dat The .dat copy opened the just-fallocated file with truncate, freeing its blocks at open and spending the headroom again as the stream re-grew it. Append mode preserves them, matching the Go side. --------- Co-authored-by: chrislusf <chrislusf@gmail.com> Co-authored-by: user.email <chrislusf@users.noreply.github.com>
690 lines
23 KiB
Go
690 lines
23 KiB
Go
package storage
|
|
|
|
import (
|
|
"errors"
|
|
"fmt"
|
|
"os"
|
|
"path"
|
|
"strconv"
|
|
"strings"
|
|
"sync"
|
|
"sync/atomic"
|
|
"time"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
|
|
"github.com/seaweedfs/seaweedfs/weed/stats"
|
|
"github.com/seaweedfs/seaweedfs/weed/storage/backend"
|
|
"github.com/seaweedfs/seaweedfs/weed/storage/needle"
|
|
"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
|
|
"github.com/seaweedfs/seaweedfs/weed/storage/types"
|
|
"github.com/seaweedfs/seaweedfs/weed/storage/volume_info"
|
|
"github.com/seaweedfs/seaweedfs/weed/util"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/glog"
|
|
)
|
|
|
|
type Volume struct {
|
|
Id needle.VolumeId
|
|
dir string
|
|
dirIdx string
|
|
Collection string
|
|
DataBackend backend.BackendStorageFile
|
|
nm NeedleMapper
|
|
tmpNm TempNeedleMapper
|
|
needleMapKind NeedleMapKind
|
|
noWriteOrDelete bool // if readonly, either noWriteOrDelete or noWriteCanDelete
|
|
noWriteCanDelete bool // if readonly, either noWriteOrDelete or noWriteCanDelete
|
|
ioUnavailable bool
|
|
ioUnavailableError string
|
|
noWriteLock sync.RWMutex
|
|
hasRemoteFile atomic.Bool // if the volume is tiered: data lives in a remote backend
|
|
MemoryMapMaxSizeMb uint32
|
|
|
|
super_block.SuperBlock
|
|
|
|
dataFileAccessLock sync.RWMutex
|
|
// Serializes a data-file mutation with the sync that makes it durable:
|
|
// write, delete, lifecycle and compaction paths hold it for their whole
|
|
// critical section, while the fsync itself runs with dataFileAccessLock
|
|
// released so readers of this volume do not wait on it.
|
|
dataFileSyncLock sync.Mutex
|
|
superBlockAccessLock sync.Mutex
|
|
|
|
// The batch worker exists only once the volume takes a durable write. Most
|
|
// never do -- read-only, remote-tiered, or written without fsync -- and a
|
|
// parked worker costs its goroutine stack plus a 128-slot channel, which a
|
|
// server holding millions of volumes cannot pay for all of them.
|
|
asyncWorkerLock sync.Mutex
|
|
asyncRequestsChan chan *needle.AsyncRequest
|
|
asyncWorkerClosed bool
|
|
|
|
lastModifiedTsSeconds uint64 // unix time in seconds
|
|
lastAppendAtNs uint64 // unix time in nanoseconds
|
|
lastWriteAppendAtNs uint64 // AppendAtNs of the newest write; tombstones don't move it
|
|
lastWriteNeedleKey types.NeedleId // the write behind the watermark
|
|
lastWriteDeleted bool // that write was deleted, so recovery has to rescan
|
|
keepLastModifiedTsOnLoad bool
|
|
|
|
lastCompactIndexOffset uint64
|
|
lastCompactRevision uint16
|
|
ldbTimeout int64
|
|
|
|
isCompactionInProgress atomic.Bool
|
|
lastDiskCheckNs atomic.Int64 // unix time in nanoseconds for phantom volume detection
|
|
|
|
volumeInfoRWLock sync.RWMutex
|
|
volumeInfo *volume_server_pb.VolumeInfo
|
|
location *DiskLocation
|
|
diskId uint32 // ID of this volume's disk in Store.Locations array
|
|
|
|
IoErrorTracker
|
|
}
|
|
|
|
func NewVolume(dirname string, dirIdx string, collection string, id needle.VolumeId, needleMapKind NeedleMapKind, replicaPlacement *super_block.ReplicaPlacement, ttl *needle.TTL, preallocate int64, ver needle.Version, memoryMapMaxSizeMb uint32, ldbTimeout int64) (v *Volume, e error) {
|
|
// if replicaPlacement is nil, the superblock will be loaded from disk
|
|
v = &Volume{dir: dirname, dirIdx: dirIdx, Collection: collection, Id: id, MemoryMapMaxSizeMb: memoryMapMaxSizeMb}
|
|
v.SuperBlock = super_block.SuperBlock{ReplicaPlacement: replicaPlacement, Ttl: ttl}
|
|
v.needleMapKind = needleMapKind
|
|
v.ldbTimeout = ldbTimeout
|
|
e = v.load(true, true, needleMapKind, preallocate, ver)
|
|
return
|
|
}
|
|
|
|
func (v *Volume) String() string {
|
|
v.noWriteLock.RLock()
|
|
defer v.noWriteLock.RUnlock()
|
|
return fmt.Sprintf("Id:%v dir:%s dirIdx:%s Collection:%s dataFile:%v nm:%v noWrite:%v canDelete:%v", v.Id, v.dir, v.dirIdx, v.Collection, v.DataBackend, v.nm, v.noWriteOrDelete || v.noWriteCanDelete, v.noWriteCanDelete)
|
|
}
|
|
|
|
func VolumeFileName(dir string, collection string, id int) (fileName string) {
|
|
idString := strconv.Itoa(id)
|
|
if collection == "" {
|
|
fileName = path.Join(dir, idString)
|
|
} else {
|
|
fileName = path.Join(dir, collection+"_"+idString)
|
|
}
|
|
return
|
|
}
|
|
|
|
func (v *Volume) DataFileName() (fileName string) {
|
|
return VolumeFileName(v.dir, v.Collection, int(v.Id))
|
|
}
|
|
|
|
func (v *Volume) IndexFileName() (fileName string) {
|
|
return VolumeFileName(v.dirIdx, v.Collection, int(v.Id))
|
|
}
|
|
|
|
func (v *Volume) FileName(ext string) (fileName string) {
|
|
switch ext {
|
|
case ".idx", ".cpx", ".ldb", ".cpldb", ".rdb":
|
|
return VolumeFileName(v.dirIdx, v.Collection, int(v.Id)) + ext
|
|
}
|
|
// .dat, .cpd, .vif
|
|
return VolumeFileName(v.dir, v.Collection, int(v.Id)) + ext
|
|
}
|
|
|
|
// RelocateIndexTo moves the volume's index to newIdxDir and reopens the volume
|
|
// against it in place, without unmounting. It takes the data-file write lock —
|
|
// so a concurrent read blocks briefly instead of failing — closes the needle
|
|
// map and data backend, moves the .idx (and the derived .sdx best-effort), then
|
|
// retargets dirIdx and reloads, mirroring CommitCompact's close-swap-load. A
|
|
// decode co-locates the rebuilt index with the data so the on-demand mount can
|
|
// find the volume; this returns it to the -dir.idx tier once the EC shards are
|
|
// gone. A no-op when the index already lives in newIdxDir. A derived .ldb is
|
|
// not moved: the reload rebuilds it in newIdxDir from the .idx.
|
|
func (v *Volume) RelocateIndexTo(newIdxDir string) error {
|
|
v.dataFileSyncLock.Lock()
|
|
defer v.dataFileSyncLock.Unlock()
|
|
v.dataFileAccessLock.Lock()
|
|
defer v.dataFileAccessLock.Unlock()
|
|
|
|
if v.dirIdx == newIdxDir {
|
|
return nil
|
|
}
|
|
oldBase := VolumeFileName(v.dirIdx, v.Collection, int(v.Id))
|
|
if _, err := os.Stat(oldBase + ".idx"); err != nil {
|
|
return nil // nothing co-located to move
|
|
}
|
|
newBase := VolumeFileName(newIdxDir, v.Collection, int(v.Id))
|
|
|
|
if v.nm != nil {
|
|
_ = v.nm.Sync()
|
|
v.nm.Close()
|
|
v.nm = nil
|
|
}
|
|
if v.DataBackend != nil {
|
|
_ = v.DataBackend.Sync()
|
|
_ = v.DataBackend.Close()
|
|
v.DataBackend = nil
|
|
}
|
|
|
|
if err := RenameOrCopyFile(oldBase+".idx", newBase+".idx"); err != nil {
|
|
// Reopen against the old dir so the volume is not left down; surface a
|
|
// failed reopen since it leaves the volume unusable until the next load.
|
|
if reopenErr := v.load(true, false, v.needleMapKind, 0, v.Version()); reopenErr != nil {
|
|
glog.Errorf("relocate volume %d: reopen after failed .idx move: %v", v.Id, reopenErr)
|
|
}
|
|
return fmt.Errorf("relocate index for volume %d: move .idx: %w", v.Id, err)
|
|
}
|
|
// The .sdx is a derived sorted index; move it when present, but a failure is
|
|
// not fatal — drop the stale copy so the reload rebuilds it in the new dir.
|
|
if _, err := os.Stat(oldBase + ".sdx"); err == nil {
|
|
if err := RenameOrCopyFile(oldBase+".sdx", newBase+".sdx"); err != nil {
|
|
glog.Warningf("relocate volume %d: move .sdx: %v (will rebuild)", v.Id, err)
|
|
_ = os.Remove(oldBase + ".sdx")
|
|
}
|
|
}
|
|
v.dirIdx = newIdxDir
|
|
return v.load(true, false, v.needleMapKind, 0, v.Version())
|
|
}
|
|
|
|
func (v *Volume) Version() needle.Version {
|
|
v.superBlockAccessLock.Lock()
|
|
defer v.superBlockAccessLock.Unlock()
|
|
if v.volumeInfo.Version != 0 {
|
|
v.SuperBlock.Version = needle.Version(v.volumeInfo.Version)
|
|
}
|
|
return v.SuperBlock.Version
|
|
}
|
|
|
|
// UpdateReplicaPlacement records rp in the live volume — the in-memory .vif
|
|
// copy and the superblock — and rewrites the .dat header when the data file
|
|
// is local. Keeps a mounted volume consistent with the .vif the caller just
|
|
// configured on disk, so heartbeats and later compactions report and rewrite
|
|
// the same placement.
|
|
func (v *Volume) UpdateReplicaPlacement(rp *super_block.ReplicaPlacement) error {
|
|
v.dataFileSyncLock.Lock()
|
|
defer v.dataFileSyncLock.Unlock()
|
|
v.dataFileAccessLock.Lock()
|
|
defer v.dataFileAccessLock.Unlock()
|
|
v.superBlockAccessLock.Lock()
|
|
defer v.superBlockAccessLock.Unlock()
|
|
v.volumeInfoRWLock.Lock()
|
|
defer v.volumeInfoRWLock.Unlock()
|
|
|
|
v.SuperBlock.ReplicaPlacement = rp
|
|
if v.volumeInfo != nil {
|
|
v.volumeInfo.Replication = rp.String()
|
|
}
|
|
diskFile, ok := v.DataBackend.(*backend.DiskFile)
|
|
if !ok {
|
|
return nil // remote-tiered: the .vif carries the placement
|
|
}
|
|
if _, err := diskFile.WriteAt(v.SuperBlock.Bytes(), 0); err != nil {
|
|
return fmt.Errorf("volume %d failed to write superblock: %v", v.Id, err)
|
|
}
|
|
return diskFile.Sync()
|
|
}
|
|
|
|
func (v *Volume) FileStat() (datSize uint64, idxSize uint64, modTime time.Time) {
|
|
v.dataFileAccessLock.RLock()
|
|
defer v.dataFileAccessLock.RUnlock()
|
|
|
|
if v.DataBackend == nil {
|
|
return
|
|
}
|
|
|
|
datFileSize, modTime, e := v.DataBackend.GetStat()
|
|
if e == nil {
|
|
return uint64(datFileSize), v.nm.IndexFileSize(), modTime
|
|
}
|
|
glog.V(0).Infof("Failed to read file size %s %v", v.DataBackend.Name(), e)
|
|
return // -1 causes integer overflow and the volume to become unwritable.
|
|
}
|
|
|
|
// datAllocatedSize returns the disk space held by the .dat file, or 0 when it is unknown.
|
|
func (v *Volume) datAllocatedSize() uint64 {
|
|
v.dataFileAccessLock.RLock()
|
|
defer v.dataFileAccessLock.RUnlock()
|
|
|
|
df, ok := v.DataBackend.(*backend.DiskFile)
|
|
if !ok {
|
|
return 0
|
|
}
|
|
fi, err := df.File.Stat()
|
|
if err != nil {
|
|
return 0
|
|
}
|
|
return AllocatedSize(fi)
|
|
}
|
|
|
|
func (v *Volume) ContentSize() uint64 {
|
|
v.dataFileAccessLock.RLock()
|
|
defer v.dataFileAccessLock.RUnlock()
|
|
if v.nm == nil {
|
|
return 0
|
|
}
|
|
return v.nm.ContentSize()
|
|
}
|
|
|
|
func (v *Volume) doIsEmpty() (bool, error) {
|
|
// check v.DataBackend.GetStat()
|
|
if v.DataBackend == nil {
|
|
return false, fmt.Errorf("v.DataBackend is nil")
|
|
} else {
|
|
datFileSize, _, e := v.DataBackend.GetStat()
|
|
if e != nil {
|
|
glog.V(0).Infof("Failed to read file size %s %v", v.DataBackend.Name(), e)
|
|
return false, fmt.Errorf("v.DataBackend.GetStat(): %v", e)
|
|
}
|
|
if datFileSize > super_block.SuperBlockSize {
|
|
return false, nil
|
|
}
|
|
}
|
|
// check v.nm.ContentSize()
|
|
if v.nm != nil {
|
|
if v.nm.ContentSize() > 0 {
|
|
return false, nil
|
|
}
|
|
}
|
|
return true, nil
|
|
}
|
|
|
|
// doIsGarbage reports whether every byte ever written is already deleted —
|
|
// the same all-garbage state vacuum measures, checked against the byte
|
|
// counters because the count counters drift on reload.
|
|
func (v *Volume) doIsGarbage() bool {
|
|
if v.nm == nil {
|
|
return false
|
|
}
|
|
contentSize := v.nm.ContentSize()
|
|
return contentSize > 0 && v.nm.DeletedSize() >= contentSize
|
|
}
|
|
|
|
func (v *Volume) DeletedSize() uint64 {
|
|
v.dataFileAccessLock.RLock()
|
|
defer v.dataFileAccessLock.RUnlock()
|
|
if v.nm == nil {
|
|
return 0
|
|
}
|
|
return v.nm.DeletedSize()
|
|
}
|
|
|
|
func (v *Volume) FileCount() uint64 {
|
|
v.dataFileAccessLock.RLock()
|
|
defer v.dataFileAccessLock.RUnlock()
|
|
if v.nm == nil {
|
|
return 0
|
|
}
|
|
return uint64(v.nm.FileCount())
|
|
}
|
|
|
|
func (v *Volume) DeletedCount() uint64 {
|
|
v.dataFileAccessLock.RLock()
|
|
defer v.dataFileAccessLock.RUnlock()
|
|
if v.nm == nil {
|
|
return 0
|
|
}
|
|
return uint64(v.nm.DeletedCount())
|
|
}
|
|
|
|
func (v *Volume) MaxFileKey() types.NeedleId {
|
|
v.dataFileAccessLock.RLock()
|
|
defer v.dataFileAccessLock.RUnlock()
|
|
if v.nm == nil {
|
|
return 0
|
|
}
|
|
return v.nm.MaxFileKey()
|
|
}
|
|
|
|
func (v *Volume) IndexFileSize() uint64 {
|
|
v.dataFileAccessLock.RLock()
|
|
defer v.dataFileAccessLock.RUnlock()
|
|
if v.nm == nil {
|
|
return 0
|
|
}
|
|
return v.nm.IndexFileSize()
|
|
}
|
|
|
|
func (v *Volume) DiskType() types.DiskType {
|
|
return v.location.DiskType
|
|
}
|
|
|
|
func (v *Volume) SyncToDisk() {
|
|
v.dataFileSyncLock.Lock()
|
|
defer v.dataFileSyncLock.Unlock()
|
|
v.dataFileAccessLock.Lock()
|
|
defer v.dataFileAccessLock.Unlock()
|
|
// Flush with the access lock dropped so readers are not stalled by the
|
|
// fsync; writers and file swaps stay excluded by dataFileSyncLock.
|
|
nm, backend := v.nm, v.DataBackend
|
|
v.dataFileAccessLock.Unlock()
|
|
if nm != nil {
|
|
if err := nm.Sync(); err != nil {
|
|
glog.Warningf("Volume Close fail to sync volume idx %d", v.Id)
|
|
}
|
|
}
|
|
if backend != nil {
|
|
if err := backend.Sync(); err != nil {
|
|
glog.Warningf("Volume Close fail to sync volume %d", v.Id)
|
|
}
|
|
}
|
|
v.dataFileAccessLock.Lock()
|
|
}
|
|
|
|
// Close cleanly shuts down this volume
|
|
func (v *Volume) Close() {
|
|
// Wait for any in-progress compaction to finish and claim the flag so no
|
|
// new compaction can start. This must happen BEFORE acquiring
|
|
// dataFileSyncLock and dataFileAccessLock to avoid deadlocking with
|
|
// CommitCompact which holds the flag while waiting for the locks.
|
|
for !v.isCompactionInProgress.CompareAndSwap(false, true) {
|
|
time.Sleep(521 * time.Millisecond)
|
|
glog.Warningf("Volume Close wait for compaction %d", v.Id)
|
|
}
|
|
defer v.isCompactionInProgress.Store(false)
|
|
|
|
v.dataFileSyncLock.Lock()
|
|
defer v.dataFileSyncLock.Unlock()
|
|
v.dataFileAccessLock.Lock()
|
|
defer v.dataFileAccessLock.Unlock()
|
|
|
|
v.doClose()
|
|
}
|
|
|
|
// SwapDataBackend atomically replaces the data backend and updates the
|
|
// remote-tier flag under dataFileAccessLock, closing the old backend. Both tier
|
|
// directions go through here so hasRemoteFile always matches the live backend:
|
|
// tier-down passes hasRemoteFile=false (now serving a local .dat), tier-up
|
|
// passes true. Keeping the swap and the flag under one lock means the heartbeat
|
|
// never observes a half-swapped backend or a flag that disagrees with it — a
|
|
// stale-false flag would make doDeleteRequest skip the new .dat's tombstones and
|
|
// disable the phantom-.dat guard.
|
|
func (v *Volume) SwapDataBackend(newBackend backend.BackendStorageFile, hasRemoteFile bool) {
|
|
v.dataFileSyncLock.Lock()
|
|
defer v.dataFileSyncLock.Unlock()
|
|
v.dataFileAccessLock.Lock()
|
|
defer v.dataFileAccessLock.Unlock()
|
|
v.swapDataBackendLocked(newBackend, hasRemoteFile)
|
|
}
|
|
|
|
// swapDataBackendLocked is the body of SwapDataBackend for callers that already
|
|
// hold dataFileAccessLock (e.g. load() reached while CommitCompact holds the
|
|
// lock). Reusing it from those under-lock paths avoids re-entering the
|
|
// non-reentrant lock, which would deadlock.
|
|
func (v *Volume) swapDataBackendLocked(newBackend backend.BackendStorageFile, hasRemoteFile bool) {
|
|
if v.DataBackend != nil {
|
|
v.DataBackend.Close()
|
|
}
|
|
v.DataBackend = newBackend
|
|
v.hasRemoteFile.Store(hasRemoteFile)
|
|
}
|
|
|
|
func (v *Volume) doClose() {
|
|
if v.nm != nil {
|
|
if err := v.nm.Sync(); err != nil {
|
|
glog.Warningf("Volume Close fail to sync volume idx %d", v.Id)
|
|
}
|
|
v.nm.Close()
|
|
v.nm = nil
|
|
}
|
|
if v.DataBackend != nil {
|
|
if err := v.DataBackend.Close(); err != nil {
|
|
glog.Warningf("Volume Close fail to sync volume %d", v.Id)
|
|
}
|
|
v.DataBackend = nil
|
|
stats.VolumeServerVolumeGauge.WithLabelValues(v.Collection, "volume").Dec()
|
|
}
|
|
}
|
|
|
|
func (v *Volume) NeedToReplicate() bool {
|
|
return v.ReplicaPlacement.GetCopyCount() > 1
|
|
}
|
|
|
|
// volume is expired if modified time + volume ttl < now
|
|
// except when volume is empty
|
|
// or when the volume does not have a ttl
|
|
// or when volumeSizeLimit is 0 when server just starts
|
|
func (v *Volume) expired(contentSize uint64, volumeSizeLimit uint64) bool {
|
|
if volumeSizeLimit == 0 {
|
|
// skip if we don't know size limit
|
|
return false
|
|
}
|
|
if contentSize <= super_block.SuperBlockSize {
|
|
return false
|
|
}
|
|
if v.Ttl == nil || v.Ttl.Minutes() == 0 {
|
|
return false
|
|
}
|
|
glog.V(2).Infof("volume %d now:%v lastModified:%v", v.Id, time.Now().Unix(), v.lastModifiedTsSeconds)
|
|
livedMinutes := (time.Now().Unix() - int64(v.lastModifiedTsSeconds)) / 60
|
|
glog.V(2).Infof("volume %d ttl:%v lived:%v", v.Id, v.Ttl, livedMinutes)
|
|
if int64(v.Ttl.Minutes()) < livedMinutes {
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
|
|
// ExpireAtSec is when this volume's data becomes garbage, counted from its last
|
|
// write. Counting from the current time instead let every .vif rewrite -- a
|
|
// read-only mark, a tier upload, an EC encode -- hand an already expiring volume
|
|
// another full TTL. Zero when the volume has no TTL.
|
|
func (v *Volume) ExpireAtSec() uint64 {
|
|
if v.Ttl == nil {
|
|
return 0
|
|
}
|
|
ttlSeconds := v.Ttl.ToSeconds()
|
|
if ttlSeconds == 0 {
|
|
return 0
|
|
}
|
|
lastWriteSec := v.lastModifiedTsSeconds
|
|
if lastWriteSec == 0 {
|
|
lastWriteSec = uint64(time.Now().Unix())
|
|
}
|
|
return lastWriteSec + ttlSeconds
|
|
}
|
|
|
|
// wait either maxDelayMinutes or 10% of ttl minutes
|
|
func (v *Volume) expiredLongEnough(maxDelayMinutes uint32) bool {
|
|
if v.Ttl == nil || v.Ttl.Minutes() == 0 {
|
|
return false
|
|
}
|
|
removalDelay := v.Ttl.Minutes() / 10
|
|
if removalDelay > maxDelayMinutes {
|
|
removalDelay = maxDelayMinutes
|
|
}
|
|
|
|
if uint64(v.Ttl.Minutes()+removalDelay)*60+v.lastModifiedTsSeconds < uint64(time.Now().Unix()) {
|
|
return true
|
|
}
|
|
return false
|
|
}
|
|
|
|
func (v *Volume) collectStatus() (maxFileKey types.NeedleId, datFileSize int64, modTime time.Time, fileCount, deletedCount, deletedSize uint64, ok bool) {
|
|
v.dataFileAccessLock.RLock()
|
|
defer v.dataFileAccessLock.RUnlock()
|
|
|
|
if v.nm == nil || v.DataBackend == nil {
|
|
return
|
|
}
|
|
|
|
ok = true
|
|
|
|
maxFileKey = v.nm.MaxFileKey()
|
|
datFileSize, modTime, _ = v.DataBackend.GetStat()
|
|
fileCount = uint64(v.nm.FileCount())
|
|
deletedCount = uint64(v.nm.DeletedCount())
|
|
deletedSize = v.nm.DeletedSize()
|
|
|
|
return
|
|
}
|
|
|
|
// ToVolumeInformationMessage fills into with what the master is told about this
|
|
// volume, allocating a message when into is nil. A heartbeat that keeps only
|
|
// the volumes it reports fills the same message for all the rest.
|
|
func (v *Volume) ToVolumeInformationMessage(into *master_pb.VolumeInformationMessage) (types.NeedleId, *master_pb.VolumeInformationMessage) {
|
|
|
|
maxFileKey, volumeSize, modTime, fileCount, deletedCount, deletedSize, ok := v.collectStatus()
|
|
|
|
if !ok {
|
|
return 0, nil
|
|
}
|
|
|
|
// 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.
|
|
if fileCount > 0 && !v.HasRemoteFile() {
|
|
const diskCheckIntervalNs = 30 * int64(time.Second)
|
|
now := time.Now().UnixNano()
|
|
if now-v.lastDiskCheckNs.Load() > diskCheckIntervalNs {
|
|
if _, err := os.Stat(v.FileName(".dat")); os.IsNotExist(err) {
|
|
glog.Warningf("Volume %d: data file %s missing (held open as deleted FD) - not reporting to master", v.Id, v.FileName(".dat"))
|
|
return 0, nil
|
|
}
|
|
v.lastDiskCheckNs.Store(now)
|
|
}
|
|
}
|
|
|
|
volumeInfo := into
|
|
if volumeInfo == nil {
|
|
volumeInfo = &master_pb.VolumeInformationMessage{}
|
|
}
|
|
volumeInfo.Id = uint32(v.Id)
|
|
volumeInfo.Size = uint64(volumeSize)
|
|
volumeInfo.Collection = v.Collection
|
|
volumeInfo.FileCount = fileCount
|
|
volumeInfo.DeleteCount = deletedCount
|
|
volumeInfo.DeletedByteCount = deletedSize
|
|
volumeInfo.ReadOnly, _, volumeInfo.ReadOnlyCanDelete, _ = v.ReadOnlyReasons()
|
|
volumeInfo.ReplicaPlacement = uint32(v.ReplicaPlacement.Byte())
|
|
volumeInfo.Version = uint32(v.Version())
|
|
volumeInfo.Ttl = v.Ttl.ToUint32()
|
|
volumeInfo.CompactRevision = uint32(v.SuperBlock.CompactionRevision)
|
|
volumeInfo.ModifiedAtSecond = modTime.Unix()
|
|
volumeInfo.DiskType = string(v.location.DiskType)
|
|
volumeInfo.DiskId = v.diskId
|
|
volumeInfo.RemoteStorageName, volumeInfo.RemoteStorageKey = v.RemoteStorageNameKey()
|
|
|
|
return maxFileKey, volumeInfo
|
|
}
|
|
|
|
func (v *Volume) RemoteStorageNameKey() (storageName, storageKey string) {
|
|
if v.volumeInfo == nil {
|
|
return
|
|
}
|
|
if len(v.volumeInfo.GetFiles()) == 0 {
|
|
return
|
|
}
|
|
return v.volumeInfo.GetFiles()[0].BackendName(), v.volumeInfo.GetFiles()[0].GetKey()
|
|
}
|
|
|
|
func (v *Volume) IsReadOnly() bool {
|
|
readOnly, _, _, _ := v.ReadOnlyReasons()
|
|
return readOnly
|
|
}
|
|
|
|
// ReadOnlyReasons reports whether the volume refuses writes and why, reading the
|
|
// flags once so the reasons cannot disagree with the verdict.
|
|
func (v *Volume) ReadOnlyReasons() (readOnly, noWriteOrDelete, noWriteCanDelete, diskSpaceLow bool) {
|
|
v.noWriteLock.RLock()
|
|
noWriteOrDelete, noWriteCanDelete = v.noWriteOrDelete, v.noWriteCanDelete
|
|
if v.ioUnavailable {
|
|
noWriteOrDelete = true
|
|
}
|
|
v.noWriteLock.RUnlock()
|
|
// The location is attached when the volume joins a disk location, which is
|
|
// after NewVolume hands it back.
|
|
diskSpaceLow = v.location != nil && v.location.isDiskSpaceLow.Load()
|
|
return noWriteOrDelete || noWriteCanDelete || diskSpaceLow, noWriteOrDelete, noWriteCanDelete, diskSpaceLow
|
|
}
|
|
|
|
var errVolumeUnavailable = errors.New("volume unavailable")
|
|
|
|
func (v *Volume) UnavailableError() error {
|
|
v.noWriteLock.RLock()
|
|
unavailable := v.ioUnavailable
|
|
reason := v.ioUnavailableError
|
|
v.noWriteLock.RUnlock()
|
|
if !unavailable {
|
|
return nil
|
|
}
|
|
if reason == "" {
|
|
return fmt.Errorf("volume %d is unavailable: %w", v.Id, errVolumeUnavailable)
|
|
}
|
|
return fmt.Errorf("volume %d is unavailable: %s: %w", v.Id, reason, errVolumeUnavailable)
|
|
}
|
|
|
|
func (v *Volume) markIoUnavailable(err error) {
|
|
v.noWriteLock.Lock()
|
|
v.noWriteOrDelete = true
|
|
v.ioUnavailable = true
|
|
v.ioUnavailableError = err.Error()
|
|
v.noWriteLock.Unlock()
|
|
v.markIoQuarantined()
|
|
|
|
if persistErr := v.persistUnavailable(err.Error()); persistErr != nil {
|
|
glog.Warningf("volume %d: failed to persist unavailable marker: %v", v.Id, persistErr)
|
|
}
|
|
if v.volumeInfo != nil {
|
|
if persistErr := v.PersistReadOnly(true, false); persistErr != nil {
|
|
glog.Warningf("volume %d: failed to persist unavailable state: %v", v.Id, persistErr)
|
|
}
|
|
}
|
|
glog.Errorf("volume %d entered unavailable state after failed recovery: %v", v.Id, err)
|
|
}
|
|
|
|
// persistUnavailable records the failed-recovery state so a reload keeps the
|
|
// volume unavailable instead of serving an unverified .dat/index pair.
|
|
func (v *Volume) persistUnavailable(reason string) error {
|
|
marker := v.FileName(".unavailable")
|
|
f, err := os.OpenFile(marker, os.O_WRONLY|os.O_CREATE|os.O_TRUNC, 0644)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if _, err := f.WriteString(reason); err != nil {
|
|
f.Close()
|
|
return err
|
|
}
|
|
if err := f.Sync(); err != nil {
|
|
f.Close()
|
|
return err
|
|
}
|
|
if err := f.Close(); err != nil {
|
|
return err
|
|
}
|
|
return util.FsyncDir(v.dir)
|
|
}
|
|
|
|
// restoreUnavailable re-arms the in-memory state the .unavailable marker
|
|
// recorded. The marker is deleted manually once the volume pair is verified.
|
|
func (v *Volume) restoreUnavailable() {
|
|
reason, err := os.ReadFile(v.FileName(".unavailable"))
|
|
if err != nil {
|
|
return
|
|
}
|
|
v.noWriteLock.Lock()
|
|
v.noWriteOrDelete = true
|
|
v.ioUnavailable = true
|
|
v.ioUnavailableError = strings.TrimSpace(string(reason))
|
|
v.noWriteLock.Unlock()
|
|
v.markIoQuarantined()
|
|
glog.Warningf("volume %d is unavailable: %s", v.Id, v.ioUnavailableError)
|
|
}
|
|
|
|
func (v *Volume) PersistReadOnly(readOnly bool, canDelete bool) error {
|
|
v.volumeInfoRWLock.Lock()
|
|
defer v.volumeInfoRWLock.Unlock()
|
|
prevReadOnly := v.volumeInfo.ReadOnly
|
|
prevReadOnlyCanDelete := v.volumeInfo.ReadOnlyCanDelete
|
|
v.volumeInfo.ReadOnly = readOnly
|
|
v.volumeInfo.ReadOnlyCanDelete = readOnly && canDelete
|
|
if err := v.SaveVolumeInfo(); err != nil {
|
|
// A pre-commit failure (write/sync/close/rename) leaves the old
|
|
// .vif intact, so roll back in-memory state to match it. A
|
|
// NotCrashDurableError means the rename already committed the
|
|
// new mode to disk; rolling back would split in-memory state
|
|
// from the durable file, so keep the new state and propagate.
|
|
var ndErr *volume_info.NotCrashDurableError
|
|
if !errors.As(err, &ndErr) {
|
|
v.volumeInfo.ReadOnly = prevReadOnly
|
|
v.volumeInfo.ReadOnlyCanDelete = prevReadOnlyCanDelete
|
|
}
|
|
return fmt.Errorf("persist volume read-only state: %w", err)
|
|
}
|
|
return nil
|
|
}
|