Merge branch 'master' of https://github.com/seaweedfs/seaweedfs
This commit is contained in:
13
.github/workflows/e2e.yml
vendored
13
.github/workflows/e2e.yml
vendored
@@ -23,14 +23,13 @@ jobs:
|
|||||||
runs-on: ubuntu-22.04
|
runs-on: ubuntu-22.04
|
||||||
timeout-minutes: 30
|
timeout-minutes: 30
|
||||||
steps:
|
steps:
|
||||||
- name: Set up Go 1.x
|
|
||||||
uses: actions/setup-go@a5f9b05d2d216f63e13859e0d847461041025775 # v2
|
|
||||||
with:
|
|
||||||
go-version: ^1.13
|
|
||||||
id: go
|
|
||||||
|
|
||||||
- name: Check out code into the Go module directory
|
- name: Check out code into the Go module directory
|
||||||
uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v2
|
uses: actions/checkout@v6
|
||||||
|
|
||||||
|
- name: Set up Go
|
||||||
|
uses: actions/setup-go@v6
|
||||||
|
with:
|
||||||
|
go-version-file: 'go.mod'
|
||||||
|
|
||||||
- name: Set up Docker Buildx
|
- name: Set up Docker Buildx
|
||||||
uses: docker/setup-buildx-action@v4
|
uses: docker/setup-buildx-action@v4
|
||||||
|
|||||||
33
.github/workflows/go.yml
vendored
33
.github/workflows/go.yml
vendored
@@ -19,13 +19,12 @@ jobs:
|
|||||||
name: Go Vet
|
name: Go Vet
|
||||||
runs-on: ubuntu-latest
|
runs-on: ubuntu-latest
|
||||||
steps:
|
steps:
|
||||||
- name: Set up Go 1.x
|
|
||||||
uses: actions/setup-go@a5f9b05d2d216f63e13859e0d847461041025775 # v2
|
|
||||||
with:
|
|
||||||
go-version: ^1.13
|
|
||||||
id: go
|
|
||||||
- name: Check out code into the Go module directory
|
- name: Check out code into the Go module directory
|
||||||
uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v2
|
uses: actions/checkout@v6
|
||||||
|
- name: Set up Go
|
||||||
|
uses: actions/setup-go@v6
|
||||||
|
with:
|
||||||
|
go-version-file: 'go.mod'
|
||||||
- name: Get dependencies
|
- name: Get dependencies
|
||||||
run: |
|
run: |
|
||||||
cd weed; go get -v -t -d ./...
|
cd weed; go get -v -t -d ./...
|
||||||
@@ -42,13 +41,12 @@ jobs:
|
|||||||
name: Build
|
name: Build
|
||||||
runs-on: ubuntu-latest
|
runs-on: ubuntu-latest
|
||||||
steps:
|
steps:
|
||||||
- name: Set up Go 1.x
|
|
||||||
uses: actions/setup-go@a5f9b05d2d216f63e13859e0d847461041025775 # v2
|
|
||||||
with:
|
|
||||||
go-version: ^1.13
|
|
||||||
id: go
|
|
||||||
- name: Check out code into the Go module directory
|
- name: Check out code into the Go module directory
|
||||||
uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v2
|
uses: actions/checkout@v6
|
||||||
|
- name: Set up Go
|
||||||
|
uses: actions/setup-go@v6
|
||||||
|
with:
|
||||||
|
go-version-file: 'go.mod'
|
||||||
- name: Build
|
- name: Build
|
||||||
run: cd weed; go build -tags "elastic gocdk sqlite ydb tarantool tikv rclone" -v .
|
run: cd weed; go build -tags "elastic gocdk sqlite ydb tarantool tikv rclone" -v .
|
||||||
|
|
||||||
@@ -56,12 +54,11 @@ jobs:
|
|||||||
name: Test
|
name: Test
|
||||||
runs-on: ubuntu-latest
|
runs-on: ubuntu-latest
|
||||||
steps:
|
steps:
|
||||||
- name: Set up Go 1.x
|
|
||||||
uses: actions/setup-go@a5f9b05d2d216f63e13859e0d847461041025775 # v2
|
|
||||||
with:
|
|
||||||
go-version: ^1.13
|
|
||||||
id: go
|
|
||||||
- name: Check out code into the Go module directory
|
- name: Check out code into the Go module directory
|
||||||
uses: actions/checkout@8e8c483db84b4bee98b60c0593521ed34d9990e8 # v2
|
uses: actions/checkout@v6
|
||||||
|
- name: Set up Go
|
||||||
|
uses: actions/setup-go@v6
|
||||||
|
with:
|
||||||
|
go-version-file: 'go.mod'
|
||||||
- name: Test
|
- name: Test
|
||||||
run: cd weed; go test -tags "elastic gocdk sqlite ydb tarantool tikv rclone" -v ./...
|
run: cd weed; go test -tags "elastic gocdk sqlite ydb tarantool tikv rclone" -v ./...
|
||||||
|
|||||||
@@ -30,7 +30,6 @@ permissions:
|
|||||||
contents: read
|
contents: read
|
||||||
|
|
||||||
env:
|
env:
|
||||||
GO_VERSION: '1.24'
|
|
||||||
TEST_TIMEOUT: '10m'
|
TEST_TIMEOUT: '10m'
|
||||||
|
|
||||||
jobs:
|
jobs:
|
||||||
@@ -43,10 +42,10 @@ jobs:
|
|||||||
- name: Checkout code
|
- name: Checkout code
|
||||||
uses: actions/checkout@v6
|
uses: actions/checkout@v6
|
||||||
|
|
||||||
- name: Set up Go ${{ env.GO_VERSION }}
|
- name: Set up Go
|
||||||
uses: actions/setup-go@v6
|
uses: actions/setup-go@v6
|
||||||
with:
|
with:
|
||||||
go-version: ${{ env.GO_VERSION }}
|
go-version-file: 'go.mod'
|
||||||
|
|
||||||
- name: Build SeaweedFS
|
- name: Build SeaweedFS
|
||||||
run: |
|
run: |
|
||||||
|
|||||||
5
.github/workflows/sftp-tests.yml
vendored
5
.github/workflows/sftp-tests.yml
vendored
@@ -24,7 +24,6 @@ permissions:
|
|||||||
contents: read
|
contents: read
|
||||||
|
|
||||||
env:
|
env:
|
||||||
GO_VERSION: '1.24'
|
|
||||||
TEST_TIMEOUT: '15m'
|
TEST_TIMEOUT: '15m'
|
||||||
|
|
||||||
jobs:
|
jobs:
|
||||||
@@ -37,10 +36,10 @@ jobs:
|
|||||||
- name: Checkout code
|
- name: Checkout code
|
||||||
uses: actions/checkout@v6
|
uses: actions/checkout@v6
|
||||||
|
|
||||||
- name: Set up Go ${{ env.GO_VERSION }}
|
- name: Set up Go
|
||||||
uses: actions/setup-go@v6
|
uses: actions/setup-go@v6
|
||||||
with:
|
with:
|
||||||
go-version: ${{ env.GO_VERSION }}
|
go-version-file: 'go.mod'
|
||||||
|
|
||||||
- name: Install dependencies
|
- name: Install dependencies
|
||||||
run: |
|
run: |
|
||||||
|
|||||||
@@ -28,7 +28,6 @@ permissions:
|
|||||||
contents: read
|
contents: read
|
||||||
|
|
||||||
env:
|
env:
|
||||||
GO_VERSION: '1.24'
|
|
||||||
TEST_TIMEOUT: '30m'
|
TEST_TIMEOUT: '30m'
|
||||||
|
|
||||||
jobs:
|
jobs:
|
||||||
@@ -46,10 +45,10 @@ jobs:
|
|||||||
- name: Checkout code
|
- name: Checkout code
|
||||||
uses: actions/checkout@v6
|
uses: actions/checkout@v6
|
||||||
|
|
||||||
- name: Set up Go ${{ env.GO_VERSION }}
|
- name: Set up Go
|
||||||
uses: actions/setup-go@v6
|
uses: actions/setup-go@v6
|
||||||
with:
|
with:
|
||||||
go-version: ${{ env.GO_VERSION }}
|
go-version-file: 'go.mod'
|
||||||
|
|
||||||
- name: Build SeaweedFS binary
|
- name: Build SeaweedFS binary
|
||||||
run: |
|
run: |
|
||||||
|
|||||||
@@ -32,6 +32,7 @@ const (
|
|||||||
defaultClusterContextTimeout = 10 * time.Second
|
defaultClusterContextTimeout = 10 * time.Second
|
||||||
defaultWaitingBacklogFloor = 8
|
defaultWaitingBacklogFloor = 8
|
||||||
defaultWaitingBacklogMultiplier = 4
|
defaultWaitingBacklogMultiplier = 4
|
||||||
|
maxEstimatedRuntimeCap = 8 * time.Hour
|
||||||
)
|
)
|
||||||
|
|
||||||
type schedulerPolicy struct {
|
type schedulerPolicy struct {
|
||||||
@@ -293,6 +294,26 @@ func (r *Plugin) runJobTypeIteration(jobType string, policy schedulerPolicy) boo
|
|||||||
|
|
||||||
r.setSchedulerLoopState(jobType, "executing")
|
r.setSchedulerLoopState(jobType, "executing")
|
||||||
|
|
||||||
|
// Scan proposals for the maximum estimated_runtime_seconds so the
|
||||||
|
// execution phase gets enough time for large jobs (e.g. vacuum on
|
||||||
|
// big volumes). If any proposal needs more time than the remaining
|
||||||
|
// JobTypeMaxRuntime, extend the execution context accordingly.
|
||||||
|
var maxEstimatedRuntime time.Duration
|
||||||
|
for _, p := range filtered {
|
||||||
|
if p.Parameters != nil {
|
||||||
|
if est, ok := p.Parameters["estimated_runtime_seconds"]; ok {
|
||||||
|
if v := est.GetInt64Value(); v > 0 {
|
||||||
|
if d := time.Duration(v) * time.Second; d > maxEstimatedRuntime {
|
||||||
|
maxEstimatedRuntime = d
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if maxEstimatedRuntime > maxEstimatedRuntimeCap {
|
||||||
|
maxEstimatedRuntime = maxEstimatedRuntimeCap
|
||||||
|
}
|
||||||
|
|
||||||
remaining = time.Until(start.Add(maxRuntime))
|
remaining = time.Until(start.Add(maxRuntime))
|
||||||
if remaining <= 0 {
|
if remaining <= 0 {
|
||||||
r.appendActivity(JobActivity{
|
r.appendActivity(JobActivity{
|
||||||
@@ -306,6 +327,17 @@ func (r *Plugin) runJobTypeIteration(jobType string, policy schedulerPolicy) boo
|
|||||||
return detected
|
return detected
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// If the longest estimated job exceeds the remaining JobTypeMaxRuntime,
|
||||||
|
// create a new execution context with enough headroom instead of using
|
||||||
|
// jobCtx which would cancel too early.
|
||||||
|
execCtx := jobCtx
|
||||||
|
execCancel := context.CancelFunc(func() {})
|
||||||
|
if maxEstimatedRuntime > 0 && maxEstimatedRuntime > remaining {
|
||||||
|
execCtx, execCancel = context.WithTimeout(context.Background(), maxEstimatedRuntime)
|
||||||
|
remaining = maxEstimatedRuntime
|
||||||
|
}
|
||||||
|
defer execCancel()
|
||||||
|
|
||||||
execPolicy := policy
|
execPolicy := policy
|
||||||
if execPolicy.ExecutionTimeout <= 0 {
|
if execPolicy.ExecutionTimeout <= 0 {
|
||||||
execPolicy.ExecutionTimeout = defaultScheduledExecutionTimeout
|
execPolicy.ExecutionTimeout = defaultScheduledExecutionTimeout
|
||||||
@@ -314,10 +346,10 @@ func (r *Plugin) runJobTypeIteration(jobType string, policy schedulerPolicy) boo
|
|||||||
execPolicy.ExecutionTimeout = remaining
|
execPolicy.ExecutionTimeout = remaining
|
||||||
}
|
}
|
||||||
|
|
||||||
successCount, errorCount, canceledCount := r.dispatchScheduledProposals(jobCtx, jobType, filtered, clusterContext, execPolicy)
|
successCount, errorCount, canceledCount := r.dispatchScheduledProposals(execCtx, jobType, filtered, clusterContext, execPolicy)
|
||||||
|
|
||||||
status := "success"
|
status := "success"
|
||||||
if jobCtx.Err() != nil {
|
if execCtx.Err() != nil {
|
||||||
status = "timeout"
|
status = "timeout"
|
||||||
} else if errorCount > 0 || canceledCount > 0 {
|
} else if errorCount > 0 || canceledCount > 0 {
|
||||||
status = "error"
|
status = "error"
|
||||||
@@ -937,7 +969,24 @@ func (r *Plugin) executeScheduledJobWithExecutor(
|
|||||||
if parent == nil {
|
if parent == nil {
|
||||||
parent = context.Background()
|
parent = context.Background()
|
||||||
}
|
}
|
||||||
execCtx, cancel := context.WithTimeout(parent, policy.ExecutionTimeout)
|
// Use the job's estimated runtime if provided and larger than the
|
||||||
|
// default execution timeout. This lets handlers like vacuum scale
|
||||||
|
// the timeout based on volume size so large volumes are not killed.
|
||||||
|
timeout := policy.ExecutionTimeout
|
||||||
|
if job.Parameters != nil {
|
||||||
|
if est, ok := job.Parameters["estimated_runtime_seconds"]; ok {
|
||||||
|
if v := est.GetInt64Value(); v > 0 {
|
||||||
|
estimated := time.Duration(v) * time.Second
|
||||||
|
if estimated > maxEstimatedRuntimeCap {
|
||||||
|
estimated = maxEstimatedRuntimeCap
|
||||||
|
}
|
||||||
|
if estimated > timeout {
|
||||||
|
timeout = estimated
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
execCtx, cancel := context.WithTimeout(parent, timeout)
|
||||||
_, err := r.executeJobWithExecutor(execCtx, executor, job, clusterContext, int32(attempt))
|
_, err := r.executeJobWithExecutor(execCtx, executor, job, clusterContext, int32(attempt))
|
||||||
cancel()
|
cancel()
|
||||||
if err == nil {
|
if err == nil {
|
||||||
|
|||||||
@@ -544,6 +544,10 @@ func buildVacuumProposal(result *workertypes.TaskDetectionResult) (*plugin_pb.Jo
|
|||||||
summary = summary + " on " + result.Server
|
summary = summary + " on " + result.Server
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// Estimate runtime: 5 min/GB (compact + commit + overhead)
|
||||||
|
volumeSizeGB := int64(result.TypedParams.VolumeSize/1024/1024/1024) + 1
|
||||||
|
estimatedRuntimeSeconds := volumeSizeGB * 5 * 60
|
||||||
|
|
||||||
return &plugin_pb.JobProposal{
|
return &plugin_pb.JobProposal{
|
||||||
ProposalId: proposalID,
|
ProposalId: proposalID,
|
||||||
DedupeKey: dedupeKey,
|
DedupeKey: dedupeKey,
|
||||||
@@ -564,6 +568,9 @@ func buildVacuumProposal(result *workertypes.TaskDetectionResult) (*plugin_pb.Jo
|
|||||||
"collection": {
|
"collection": {
|
||||||
Kind: &plugin_pb.ConfigValue_StringValue{StringValue: result.Collection},
|
Kind: &plugin_pb.ConfigValue_StringValue{StringValue: result.Collection},
|
||||||
},
|
},
|
||||||
|
"estimated_runtime_seconds": {
|
||||||
|
Kind: &plugin_pb.ConfigValue_Int64Value{Int64Value: estimatedRuntimeSeconds},
|
||||||
|
},
|
||||||
},
|
},
|
||||||
Labels: map[string]string{
|
Labels: map[string]string{
|
||||||
"task_type": "vacuum",
|
"task_type": "vacuum",
|
||||||
|
|||||||
@@ -300,9 +300,13 @@ func (l *DiskLocation) DeleteCollectionFromDiskLocation(collection string) (e er
|
|||||||
var wg sync.WaitGroup
|
var wg sync.WaitGroup
|
||||||
wg.Add(2)
|
wg.Add(2)
|
||||||
go func() {
|
go func() {
|
||||||
for _, v := range delVolsMap {
|
for k, v := range delVolsMap {
|
||||||
if err := v.Destroy(false); err != nil {
|
if err := v.Destroy(false); err != nil {
|
||||||
errChain <- err
|
errChain <- err
|
||||||
|
} else {
|
||||||
|
l.volumesLock.Lock()
|
||||||
|
delete(l.volumes, k)
|
||||||
|
l.volumesLock.Unlock()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
wg.Done()
|
wg.Done()
|
||||||
@@ -383,14 +387,10 @@ func (l *DiskLocation) UnloadVolume(vid needle.VolumeId) error {
|
|||||||
func (l *DiskLocation) unmountVolumeByCollection(collectionName string) map[needle.VolumeId]*Volume {
|
func (l *DiskLocation) unmountVolumeByCollection(collectionName string) map[needle.VolumeId]*Volume {
|
||||||
deltaVols := make(map[needle.VolumeId]*Volume, 0)
|
deltaVols := make(map[needle.VolumeId]*Volume, 0)
|
||||||
for k, v := range l.volumes {
|
for k, v := range l.volumes {
|
||||||
if v.Collection == collectionName && !v.isCompacting && !v.isCommitCompacting {
|
if v.Collection == collectionName && !v.isCompactionInProgress.Load() {
|
||||||
deltaVols[k] = v
|
deltaVols[k] = v
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
for k := range deltaVols {
|
|
||||||
delete(l.volumes, k)
|
|
||||||
}
|
|
||||||
return deltaVols
|
return deltaVols
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|||||||
@@ -5,6 +5,7 @@ import (
|
|||||||
"path"
|
"path"
|
||||||
"strconv"
|
"strconv"
|
||||||
"sync"
|
"sync"
|
||||||
|
"sync/atomic"
|
||||||
"time"
|
"time"
|
||||||
|
|
||||||
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
|
"github.com/seaweedfs/seaweedfs/weed/pb/master_pb"
|
||||||
@@ -45,8 +46,7 @@ type Volume struct {
|
|||||||
lastCompactRevision uint16
|
lastCompactRevision uint16
|
||||||
ldbTimeout int64
|
ldbTimeout int64
|
||||||
|
|
||||||
isCompacting bool
|
isCompactionInProgress atomic.Bool
|
||||||
isCommitCompacting bool
|
|
||||||
|
|
||||||
volumeInfoRWLock sync.RWMutex
|
volumeInfoRWLock sync.RWMutex
|
||||||
volumeInfo *volume_server_pb.VolumeInfo
|
volumeInfo *volume_server_pb.VolumeInfo
|
||||||
@@ -224,6 +224,16 @@ func (v *Volume) SyncToDisk() {
|
|||||||
|
|
||||||
// Close cleanly shuts down this volume
|
// Close cleanly shuts down this volume
|
||||||
func (v *Volume) Close() {
|
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
|
||||||
|
// dataFileAccessLock to avoid deadlocking with CommitCompact which holds
|
||||||
|
// the flag while waiting for the lock.
|
||||||
|
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.dataFileAccessLock.Lock()
|
v.dataFileAccessLock.Lock()
|
||||||
defer v.dataFileAccessLock.Unlock()
|
defer v.dataFileAccessLock.Unlock()
|
||||||
|
|
||||||
@@ -231,11 +241,6 @@ func (v *Volume) Close() {
|
|||||||
}
|
}
|
||||||
|
|
||||||
func (v *Volume) doClose() {
|
func (v *Volume) doClose() {
|
||||||
for v.isCommitCompacting {
|
|
||||||
time.Sleep(521 * time.Millisecond)
|
|
||||||
glog.Warningf("Volume Close wait for compaction %d", v.Id)
|
|
||||||
}
|
|
||||||
|
|
||||||
if v.nm != nil {
|
if v.nm != nil {
|
||||||
if err := v.nm.Sync(); err != nil {
|
if err := v.nm.Sync(); err != nil {
|
||||||
glog.Warningf("Volume Close fail to sync volume idx %d", v.Id)
|
glog.Warningf("Volume Close fail to sync volume idx %d", v.Id)
|
||||||
|
|||||||
@@ -64,22 +64,26 @@ func (v *Volume) CompactByVolumeData(opts *CompactOptions) error {
|
|||||||
//v.accessLock.Lock()
|
//v.accessLock.Lock()
|
||||||
//defer v.accessLock.Unlock()
|
//defer v.accessLock.Unlock()
|
||||||
//glog.V(3).Infof("Got Compaction lock...")
|
//glog.V(3).Infof("Got Compaction lock...")
|
||||||
if v.isCompacting || v.isCommitCompacting {
|
if !v.isCompactionInProgress.CompareAndSwap(false, true) {
|
||||||
glog.V(0).Infof("Volume %d is already compacting...", v.Id)
|
glog.V(0).Infof("Volume %d is already compacting...", v.Id)
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
v.isCompacting = true
|
defer v.isCompactionInProgress.Store(false)
|
||||||
defer func() {
|
|
||||||
v.isCompacting = false
|
|
||||||
}()
|
|
||||||
|
|
||||||
v.lastCompactIndexOffset = v.IndexFileSize()
|
v.lastCompactIndexOffset = v.IndexFileSize()
|
||||||
v.lastCompactRevision = v.SuperBlock.CompactionRevision
|
v.lastCompactRevision = v.SuperBlock.CompactionRevision
|
||||||
glog.V(3).Infof("creating copies for volume %d ,last offset %d...", v.Id, v.lastCompactIndexOffset)
|
glog.V(3).Infof("creating copies for volume %d ,last offset %d...", v.Id, v.lastCompactIndexOffset)
|
||||||
|
if v.DataBackend == nil {
|
||||||
|
return fmt.Errorf("volume %d backend is empty remote:%v", v.Id, v.HasRemoteFile())
|
||||||
|
}
|
||||||
|
nm := v.nm
|
||||||
|
if nm == nil {
|
||||||
|
return fmt.Errorf("volume %d needle map is nil", v.Id)
|
||||||
|
}
|
||||||
if err := v.DataBackend.Sync(); err != nil {
|
if err := v.DataBackend.Sync(); err != nil {
|
||||||
glog.V(0).Infof("compact failed to sync volume %d", v.Id)
|
glog.V(0).Infof("compact failed to sync volume %d", v.Id)
|
||||||
}
|
}
|
||||||
if err := v.nm.Sync(); err != nil {
|
if err := nm.Sync(); err != nil {
|
||||||
glog.V(0).Infof("compact failed to sync volume idx %d", v.Id)
|
glog.V(0).Infof("compact failed to sync volume idx %d", v.Id)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -102,14 +106,11 @@ func (v *Volume) CompactByIndex(opts *CompactOptions) error {
|
|||||||
}
|
}
|
||||||
glog.V(3).Infof("Compact2 volume %d ...", v.Id)
|
glog.V(3).Infof("Compact2 volume %d ...", v.Id)
|
||||||
|
|
||||||
if v.isCompacting || v.isCommitCompacting {
|
if !v.isCompactionInProgress.CompareAndSwap(false, true) {
|
||||||
glog.V(0).Infof("Volume %d is already compacting2 ...", v.Id)
|
glog.V(0).Infof("Volume %d is already compacting2 ...", v.Id)
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
v.isCompacting = true
|
defer v.isCompactionInProgress.Store(false)
|
||||||
defer func() {
|
|
||||||
v.isCompacting = false
|
|
||||||
}()
|
|
||||||
|
|
||||||
v.lastCompactIndexOffset = v.IndexFileSize()
|
v.lastCompactIndexOffset = v.IndexFileSize()
|
||||||
v.lastCompactRevision = v.SuperBlock.CompactionRevision
|
v.lastCompactRevision = v.SuperBlock.CompactionRevision
|
||||||
@@ -117,10 +118,14 @@ func (v *Volume) CompactByIndex(opts *CompactOptions) error {
|
|||||||
if v.DataBackend == nil {
|
if v.DataBackend == nil {
|
||||||
return fmt.Errorf("volume %d backend is empty remote:%v", v.Id, v.HasRemoteFile())
|
return fmt.Errorf("volume %d backend is empty remote:%v", v.Id, v.HasRemoteFile())
|
||||||
}
|
}
|
||||||
|
nm := v.nm
|
||||||
|
if nm == nil {
|
||||||
|
return fmt.Errorf("volume %d needle map is nil", v.Id)
|
||||||
|
}
|
||||||
if err := v.DataBackend.Sync(); err != nil {
|
if err := v.DataBackend.Sync(); err != nil {
|
||||||
glog.V(0).Infof("compact2 failed to sync volume dat %d: %v", v.Id, err)
|
glog.V(0).Infof("compact2 failed to sync volume dat %d: %v", v.Id, err)
|
||||||
}
|
}
|
||||||
if err := v.nm.Sync(); err != nil {
|
if err := nm.Sync(); err != nil {
|
||||||
glog.V(0).Infof("compact2 failed to sync volume idx %d: %v", v.Id, err)
|
glog.V(0).Infof("compact2 failed to sync volume idx %d: %v", v.Id, err)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -139,14 +144,11 @@ func (v *Volume) CommitCompact() error {
|
|||||||
}
|
}
|
||||||
glog.V(0).Infof("Committing volume %d vacuuming...", v.Id)
|
glog.V(0).Infof("Committing volume %d vacuuming...", v.Id)
|
||||||
|
|
||||||
if v.isCommitCompacting {
|
if !v.isCompactionInProgress.CompareAndSwap(false, true) {
|
||||||
glog.V(0).Infof("Volume %d is already commit compacting ...", v.Id)
|
glog.V(0).Infof("Volume %d is already compacting ...", v.Id)
|
||||||
return nil
|
return nil
|
||||||
}
|
}
|
||||||
v.isCommitCompacting = true
|
defer v.isCompactionInProgress.Store(false)
|
||||||
defer func() {
|
|
||||||
v.isCommitCompacting = false
|
|
||||||
}()
|
|
||||||
|
|
||||||
v.dataFileAccessLock.Lock()
|
v.dataFileAccessLock.Lock()
|
||||||
defer v.dataFileAccessLock.Unlock()
|
defer v.dataFileAccessLock.Unlock()
|
||||||
@@ -530,7 +532,7 @@ func (v *Volume) copyDataBasedOnIndexFile(opts *CompactOptions) (err error) {
|
|||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
}
|
}
|
||||||
if v.Ttl.String() == "" {
|
if v.Ttl.String() == "" && v.nm != nil {
|
||||||
dstDatSize, _, err := dstDatBackend.GetStat()
|
dstDatSize, _, err := dstDatBackend.GetStat()
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return err
|
return err
|
||||||
|
|||||||
@@ -70,7 +70,7 @@ func (v *Volume) Destroy(onlyEmpty bool) (err error) {
|
|||||||
return
|
return
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
if v.isCompacting || v.isCommitCompacting {
|
if !v.isCompactionInProgress.CompareAndSwap(false, true) {
|
||||||
err = fmt.Errorf("volume %d is compacting", v.Id)
|
err = fmt.Errorf("volume %d is compacting", v.Id)
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -25,6 +25,7 @@ type VacuumTask struct {
|
|||||||
garbageThreshold float64
|
garbageThreshold float64
|
||||||
progress float64
|
progress float64
|
||||||
grpcDialOption grpc.DialOption
|
grpcDialOption grpc.DialOption
|
||||||
|
volumeSize uint64
|
||||||
}
|
}
|
||||||
|
|
||||||
// NewVacuumTask creates a new unified vacuum task instance
|
// NewVacuumTask creates a new unified vacuum task instance
|
||||||
@@ -51,6 +52,7 @@ func (t *VacuumTask) Execute(ctx context.Context, params *worker_pb.TaskParams)
|
|||||||
}
|
}
|
||||||
|
|
||||||
t.garbageThreshold = vacuumParams.GarbageThreshold
|
t.garbageThreshold = vacuumParams.GarbageThreshold
|
||||||
|
t.volumeSize = params.VolumeSize
|
||||||
|
|
||||||
t.GetLogger().WithFields(map[string]interface{}{
|
t.GetLogger().WithFields(map[string]interface{}{
|
||||||
"volume_id": t.volumeID,
|
"volume_id": t.volumeID,
|
||||||
@@ -62,7 +64,7 @@ func (t *VacuumTask) Execute(ctx context.Context, params *worker_pb.TaskParams)
|
|||||||
// Step 1: Check volume status and garbage ratio
|
// Step 1: Check volume status and garbage ratio
|
||||||
t.ReportProgress(10.0)
|
t.ReportProgress(10.0)
|
||||||
t.GetLogger().Info("Checking volume status")
|
t.GetLogger().Info("Checking volume status")
|
||||||
eligible, currentGarbageRatio, err := t.checkVacuumEligibility()
|
eligible, currentGarbageRatio, err := t.checkVacuumEligibility(ctx)
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("failed to check vacuum eligibility: %v", err)
|
return fmt.Errorf("failed to check vacuum eligibility: %v", err)
|
||||||
}
|
}
|
||||||
@@ -83,14 +85,14 @@ func (t *VacuumTask) Execute(ctx context.Context, params *worker_pb.TaskParams)
|
|||||||
"threshold": t.garbageThreshold,
|
"threshold": t.garbageThreshold,
|
||||||
}).Info("Performing vacuum operation")
|
}).Info("Performing vacuum operation")
|
||||||
|
|
||||||
if err := t.performVacuum(); err != nil {
|
if err := t.performVacuum(ctx); err != nil {
|
||||||
return fmt.Errorf("failed to perform vacuum: %v", err)
|
return fmt.Errorf("failed to perform vacuum: %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Step 3: Verify vacuum results
|
// Step 3: Verify vacuum results
|
||||||
t.ReportProgress(90.0)
|
t.ReportProgress(90.0)
|
||||||
t.GetLogger().Info("Verifying vacuum results")
|
t.GetLogger().Info("Verifying vacuum results")
|
||||||
if err := t.verifyVacuumResults(); err != nil {
|
if err := t.verifyVacuumResults(ctx); err != nil {
|
||||||
glog.Warningf("Vacuum verification failed: %v", err)
|
glog.Warningf("Vacuum verification failed: %v", err)
|
||||||
// Don't fail the task - vacuum operation itself succeeded
|
// Don't fail the task - vacuum operation itself succeeded
|
||||||
}
|
}
|
||||||
@@ -146,15 +148,28 @@ func (t *VacuumTask) GetProgress() float64 {
|
|||||||
return t.progress
|
return t.progress
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// vacuumTimeout returns a dynamic timeout scaled by volume size, matching the
|
||||||
|
// topology vacuum approach. base is the per-GB multiplier (e.g. 1 minute for
|
||||||
|
// check, 3 minutes for compact).
|
||||||
|
func (t *VacuumTask) vacuumTimeout(base time.Duration) time.Duration {
|
||||||
|
if t.volumeSize == 0 {
|
||||||
|
glog.V(1).Infof("volume %d has no size metric, using minimum timeout", t.volumeID)
|
||||||
|
}
|
||||||
|
sizeGB := int64(t.volumeSize/1024/1024/1024) + 1
|
||||||
|
return base * time.Duration(sizeGB)
|
||||||
|
}
|
||||||
|
|
||||||
// Helper methods for real vacuum operations
|
// Helper methods for real vacuum operations
|
||||||
|
|
||||||
// checkVacuumEligibility checks if the volume meets vacuum criteria
|
// checkVacuumEligibility checks if the volume meets vacuum criteria
|
||||||
func (t *VacuumTask) checkVacuumEligibility() (bool, float64, error) {
|
func (t *VacuumTask) checkVacuumEligibility(ctx context.Context) (bool, float64, error) {
|
||||||
var garbageRatio float64
|
var garbageRatio float64
|
||||||
|
|
||||||
err := operation.WithVolumeServerClient(false, pb.ServerAddress(t.server), t.grpcDialOption,
|
err := operation.WithVolumeServerClient(false, pb.ServerAddress(t.server), t.grpcDialOption,
|
||||||
func(client volume_server_pb.VolumeServerClient) error {
|
func(client volume_server_pb.VolumeServerClient) error {
|
||||||
resp, err := client.VacuumVolumeCheck(context.Background(), &volume_server_pb.VacuumVolumeCheckRequest{
|
checkCtx, cancel := context.WithTimeout(ctx, t.vacuumTimeout(time.Minute))
|
||||||
|
defer cancel()
|
||||||
|
resp, err := client.VacuumVolumeCheck(checkCtx, &volume_server_pb.VacuumVolumeCheckRequest{
|
||||||
VolumeId: t.volumeID,
|
VolumeId: t.volumeID,
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -178,12 +193,14 @@ func (t *VacuumTask) checkVacuumEligibility() (bool, float64, error) {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// performVacuum executes the actual vacuum operation
|
// performVacuum executes the actual vacuum operation
|
||||||
func (t *VacuumTask) performVacuum() error {
|
func (t *VacuumTask) performVacuum(ctx context.Context) error {
|
||||||
return operation.WithVolumeServerClient(false, pb.ServerAddress(t.server), t.grpcDialOption,
|
return operation.WithVolumeServerClient(false, pb.ServerAddress(t.server), t.grpcDialOption,
|
||||||
func(client volume_server_pb.VolumeServerClient) error {
|
func(client volume_server_pb.VolumeServerClient) error {
|
||||||
// Step 1: Compact the volume
|
// Step 1: Compact the volume (3 min per GB, matching topology vacuum)
|
||||||
t.GetLogger().Info("Compacting volume")
|
t.GetLogger().Info("Compacting volume")
|
||||||
stream, err := client.VacuumVolumeCompact(context.Background(), &volume_server_pb.VacuumVolumeCompactRequest{
|
compactCtx, compactCancel := context.WithTimeout(ctx, t.vacuumTimeout(3*time.Minute))
|
||||||
|
defer compactCancel()
|
||||||
|
stream, err := client.VacuumVolumeCompact(compactCtx, &volume_server_pb.VacuumVolumeCompactRequest{
|
||||||
VolumeId: t.volumeID,
|
VolumeId: t.volumeID,
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -202,18 +219,22 @@ func (t *VacuumTask) performVacuum() error {
|
|||||||
glog.V(2).Infof("Volume %d compact progress: %d bytes processed", t.volumeID, resp.ProcessedBytes)
|
glog.V(2).Infof("Volume %d compact progress: %d bytes processed", t.volumeID, resp.ProcessedBytes)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Step 2: Commit the vacuum
|
// Step 2: Commit the vacuum (1 min per GB)
|
||||||
t.GetLogger().Info("Committing vacuum operation")
|
t.GetLogger().Info("Committing vacuum operation")
|
||||||
_, err = client.VacuumVolumeCommit(context.Background(), &volume_server_pb.VacuumVolumeCommitRequest{
|
commitCtx, commitCancel := context.WithTimeout(ctx, t.vacuumTimeout(time.Minute))
|
||||||
|
defer commitCancel()
|
||||||
|
_, err = client.VacuumVolumeCommit(commitCtx, &volume_server_pb.VacuumVolumeCommitRequest{
|
||||||
VolumeId: t.volumeID,
|
VolumeId: t.volumeID,
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
return fmt.Errorf("vacuum commit failed: %v", err)
|
return fmt.Errorf("vacuum commit failed: %v", err)
|
||||||
}
|
}
|
||||||
|
|
||||||
// Step 3: Cleanup old files
|
// Step 3: Cleanup old files (1 min per GB)
|
||||||
t.GetLogger().Info("Cleaning up vacuum files")
|
t.GetLogger().Info("Cleaning up vacuum files")
|
||||||
_, err = client.VacuumVolumeCleanup(context.Background(), &volume_server_pb.VacuumVolumeCleanupRequest{
|
cleanupCtx, cleanupCancel := context.WithTimeout(ctx, t.vacuumTimeout(time.Minute))
|
||||||
|
defer cleanupCancel()
|
||||||
|
_, err = client.VacuumVolumeCleanup(cleanupCtx, &volume_server_pb.VacuumVolumeCleanupRequest{
|
||||||
VolumeId: t.volumeID,
|
VolumeId: t.volumeID,
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
@@ -226,10 +247,12 @@ func (t *VacuumTask) performVacuum() error {
|
|||||||
}
|
}
|
||||||
|
|
||||||
// verifyVacuumResults checks the volume status after vacuum
|
// verifyVacuumResults checks the volume status after vacuum
|
||||||
func (t *VacuumTask) verifyVacuumResults() error {
|
func (t *VacuumTask) verifyVacuumResults(ctx context.Context) error {
|
||||||
return operation.WithVolumeServerClient(false, pb.ServerAddress(t.server), t.grpcDialOption,
|
return operation.WithVolumeServerClient(false, pb.ServerAddress(t.server), t.grpcDialOption,
|
||||||
func(client volume_server_pb.VolumeServerClient) error {
|
func(client volume_server_pb.VolumeServerClient) error {
|
||||||
resp, err := client.VacuumVolumeCheck(context.Background(), &volume_server_pb.VacuumVolumeCheckRequest{
|
verifyCtx, cancel := context.WithTimeout(ctx, t.vacuumTimeout(time.Minute))
|
||||||
|
defer cancel()
|
||||||
|
resp, err := client.VacuumVolumeCheck(verifyCtx, &volume_server_pb.VacuumVolumeCheckRequest{
|
||||||
VolumeId: t.volumeID,
|
VolumeId: t.volumeID,
|
||||||
})
|
})
|
||||||
if err != nil {
|
if err != nil {
|
||||||
|
|||||||
Reference in New Issue
Block a user