Files
seaweedfs/weed/storage/volume.go
T
0d721fccee fix(volume): keep preallocated growth above -volume.minFreeSpace (#12067)
* 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>
2026-10-26 08:03:25 -07:00

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
}