diff --git a/REST_API.md b/REST_API.md index a8315a5fd5..97df87757c 100644 --- a/REST_API.md +++ b/REST_API.md @@ -271,6 +271,10 @@ curl "http://localhost:9333/vol/vacuum?garbageThreshold=0.4" Vacuuming makes a volume read-only, copies live needles to a new volume, and swaps it in. +The master runs one vacuum at a time. A request that arrives while the +automatic vacuum or another request is running does nothing and gets +`409 Conflict` with `{"error":"another vacuum is running, try again later"}`. + ### Delete a collection ```bash diff --git a/test/s3/versioning/s3_copy_versioning_regression_test.go b/test/s3/versioning/s3_copy_versioning_regression_test.go index 92bbc7864b..1e2a60974f 100644 --- a/test/s3/versioning/s3_copy_versioning_regression_test.go +++ b/test/s3/versioning/s3_copy_versioning_regression_test.go @@ -44,12 +44,22 @@ func vacuumVolumes(t *testing.T) { require.NotEmpty(t, defaultConfig.MasterEndpoint, "vacuum needs a master endpoint; set MASTER_ENDPOINT") endpoint := strings.TrimRight(defaultConfig.MasterEndpoint, "/") + "/vol/vacuum?garbageThreshold=0.001" httpClient := &http.Client{Timeout: 30 * time.Second} - resp, err := httpClient.Get(endpoint) - require.NoError(t, err, "vacuum request to %s", endpoint) - defer resp.Body.Close() - _, err = io.Copy(io.Discard, resp.Body) - require.NoError(t, err, "reading the vacuum response") - require.Equal(t, http.StatusOK, resp.StatusCode, "vacuum request to %s", endpoint) + // The master runs one vacuum at a time and answers 409 to a request it + // dropped because its automatic vacuum was running. Ask until ours runs. + deadline := time.Now().Add(time.Minute) + for { + resp, err := httpClient.Get(endpoint) + require.NoError(t, err, "vacuum request to %s", endpoint) + _, err = io.Copy(io.Discard, resp.Body) + resp.Body.Close() + require.NoError(t, err, "reading the vacuum response") + if resp.StatusCode == http.StatusConflict && time.Now().Before(deadline) { + time.Sleep(500 * time.Millisecond) + continue + } + require.Equal(t, http.StatusOK, resp.StatusCode, "vacuum request to %s", endpoint) + return + } } func requireVersionBody(t *testing.T, client *s3.Client, bucketName, objectKey, versionId string, want []byte, msg string) { diff --git a/test/tus/tus_integration_test.go b/test/tus/tus_integration_test.go index cf9456325c..1cc3390a93 100644 --- a/test/tus/tus_integration_test.go +++ b/test/tus/tus_integration_test.go @@ -983,11 +983,21 @@ func TestTusAbortedPatchKeepsStoredChunks(t *testing.T) { require.Equal(t, http.StatusNoContent, patchResp.StatusCode) // A vacuum reclaims whatever the filer deleted, so the file survives this - // only if the chunks the session kept are still stored. - vacuumResp, err := client.Get(fmt.Sprintf("http://127.0.0.1:%s/vol/vacuum?garbageThreshold=0.001", testMasterPort)) - require.NoError(t, err) - vacuumResp.Body.Close() - require.Equal(t, http.StatusOK, vacuumResp.StatusCode) + // only if the chunks the session kept are still stored. The master answers + // 409 to a request it dropped because its automatic vacuum was running, so + // ask until ours runs. + vacuumDeadline := time.Now().Add(time.Minute) + for { + vacuumResp, err := client.Get(fmt.Sprintf("http://127.0.0.1:%s/vol/vacuum?garbageThreshold=0.001", testMasterPort)) + require.NoError(t, err) + vacuumResp.Body.Close() + if vacuumResp.StatusCode == http.StatusConflict && time.Now().Before(vacuumDeadline) { + time.Sleep(500 * time.Millisecond) + continue + } + require.Equal(t, http.StatusOK, vacuumResp.StatusCode) + break + } getResp, err := client.Get(cluster.FilerURL() + targetPath) require.NoError(t, err) diff --git a/weed/server/master_grpc_server_volume.go b/weed/server/master_grpc_server_volume.go index 96b60182f9..099541c583 100644 --- a/weed/server/master_grpc_server_volume.go +++ b/weed/server/master_grpc_server_volume.go @@ -407,7 +407,11 @@ func (ms *MasterServer) VacuumVolume(ctx context.Context, req *master_pb.VacuumV resp := &master_pb.VacuumVolumeResponse{} - ms.Topo.Vacuum(ms.grpcDialOption, float64(req.GarbageThreshold), ms.option.MaxParallelVacuumPerServer, ms.option.MaxParallelVacuumPerRack, req.VolumeId, req.Collection, ms.preallocateSize, false, 0) + // A request that lost the vacuum guard to the automatic vacuum or to another + // request did nothing. Returning success would tell the caller it ran. + if err := ms.Topo.Vacuum(ms.grpcDialOption, float64(req.GarbageThreshold), ms.option.MaxParallelVacuumPerServer, ms.option.MaxParallelVacuumPerRack, req.VolumeId, req.Collection, ms.preallocateSize, false, 0); err != nil { + return nil, err + } return resp, nil } diff --git a/weed/server/master_server_handlers_admin.go b/weed/server/master_server_handlers_admin.go index e6479ea49a..3bd8d4c863 100644 --- a/weed/server/master_server_handlers_admin.go +++ b/weed/server/master_server_handlers_admin.go @@ -61,7 +61,10 @@ func (ms *MasterServer) volumeVacuumHandler(w http.ResponseWriter, r *http.Reque } } // glog.Infoln("garbageThreshold =", gcThreshold) - ms.Topo.Vacuum(ms.grpcDialOption, gcThreshold, ms.option.MaxParallelVacuumPerServer, ms.option.MaxParallelVacuumPerRack, 0, "", ms.preallocateSize, false, 0) + if err := ms.Topo.Vacuum(ms.grpcDialOption, gcThreshold, ms.option.MaxParallelVacuumPerServer, ms.option.MaxParallelVacuumPerRack, 0, "", ms.preallocateSize, false, 0); err != nil { + writeJsonError(w, r, http.StatusConflict, err) + return + } ms.dirStatusHandler(w, r) } diff --git a/weed/shell/command_volume_vacuum.go b/weed/shell/command_volume_vacuum.go index cd93dcddd5..965f7e3b99 100644 --- a/weed/shell/command_volume_vacuum.go +++ b/weed/shell/command_volume_vacuum.go @@ -38,6 +38,10 @@ func (c *commandVacuum) Help() string { Without -volumeId this runs the same sweep as the automatic vacuum, which skips read-only volumes. Name a read-only volume with -volumeId to vacuum it anyway. + The master runs one vacuum at a time. While the automatic vacuum or another + request is running, this command fails instead of waiting for its turn. Run it + again later, or pause the automatic vacuum first with volume.vacuum.disable. + ` } @@ -157,13 +161,27 @@ func (c *commandVacuum) Do(args []string, commandEnv *CommandEnv, writer io.Writ return err }) if err != nil { - return err + // A list of volumes or collections is several requests, and a vacuum + // that starts in between only fails the ones after it. + return fmt.Errorf("vacuum %s: %w", describeVacuumRequest(req), err) } } return nil } +func describeVacuumRequest(req *master_pb.VacuumVolumeRequest) string { + switch { + case req.VolumeId != 0 && req.Collection != "": + return fmt.Sprintf("volume %d in collection %s", req.VolumeId, req.Collection) + case req.VolumeId != 0: + return fmt.Sprintf("volume %d", req.VolumeId) + case req.Collection != "": + return "collection " + req.Collection + } + return "all volumes" +} + // vidReplica pairs a volume replica's read-only state with the server that // reports it, so the unnamed-collection sweep can reproduce the master's // low-disk-space exception client-side. diff --git a/weed/topology/topology_vacuum.go b/weed/topology/topology_vacuum.go index 83144cea1d..6428e0259e 100644 --- a/weed/topology/topology_vacuum.go +++ b/weed/topology/topology_vacuum.go @@ -2,6 +2,7 @@ package topology import ( "context" + "errors" "io" "sync" "sync/atomic" @@ -255,13 +256,22 @@ func (t *Topology) batchVacuumVolumeCleanup(grpcDialOption grpc.DialOption, vl * } } -func (t *Topology) Vacuum(grpcDialOption grpc.DialOption, garbageThreshold float64, maxParallelVacuumPerServer int, maxParallelVacuumPerRack int, volumeId uint32, collection string, preallocate int64, automatic bool, deleteEmptyAfter time.Duration) { +// ErrVacuumRunning is what Vacuum returns when another vacuum holds the +// topology-wide guard. The request was dropped, not queued: nothing waits for +// the running vacuum, so the caller has to ask again. +var ErrVacuumRunning = errors.New("another vacuum is running, try again later") + +// Vacuum runs one vacuum at a time. A call that finds another one running, +// the automatic vacuum or an earlier request, does nothing and returns +// ErrVacuumRunning, so that a request for a volume cannot look done when it +// never started. +func (t *Topology) Vacuum(grpcDialOption grpc.DialOption, garbageThreshold float64, maxParallelVacuumPerServer int, maxParallelVacuumPerRack int, volumeId uint32, collection string, preallocate int64, automatic bool, deleteEmptyAfter time.Duration) error { // if there is vacuum going on, return immediately swapped := atomic.CompareAndSwapInt64(&t.vacuumLockCounter, 0, 1) if !swapped { glog.V(0).Infof("Vacuum is already running") - return + return ErrVacuumRunning } defer atomic.StoreInt64(&t.vacuumLockCounter, 0) @@ -301,6 +311,7 @@ func (t *Topology) Vacuum(grpcDialOption grpc.DialOption, garbageThreshold float break } } + return nil } func (t *Topology) vacuumOneVolumeLayout(grpcDialOption grpc.DialOption, volumeLayout *VolumeLayout, c *Collection, garbageThreshold float64, maxParallelVacuumPerServer int, maxParallelVacuumPerRack int, preallocate int64, automatic bool, deleteEmptyAfter time.Duration) { diff --git a/weed/topology/topology_vacuum_test.go b/weed/topology/topology_vacuum_test.go index 106011d594..0b909f1954 100644 --- a/weed/topology/topology_vacuum_test.go +++ b/weed/topology/topology_vacuum_test.go @@ -518,6 +518,83 @@ func TestVacuumStalledVolumeServerReleasesGuard(t *testing.T) { } } +// A request that finds another vacuum running is dropped, not queued. It has +// to say so: returning as if it had run leaves the caller believing its volume +// was compacted. +func TestVacuumReportsAnotherVacuumRunning(t *testing.T) { + fake := &fakeVacuumServer{ + checks: map[uint32]*volume_server_pb.VacuumVolumeCheckResponse{ + 1: {GarbageRatio: 0.9}, + 2: {GarbageRatio: 0.9}, + }, + entered: make(chan string, 1), + compactGate: make(chan struct{}), + compactTracker: &compactTracker{}, + } + grpcPort, dialOption := startFakeVolumeServer(t, fake) + + topo := NewTopology("weedfs", sequence.NewMemorySequencer(), 32*1024, 5, false) + dn := topo.GetOrCreateDataCenter("dc1").GetOrCreateRack("rack1"). + GetOrCreateDataNode("127.0.0.1", 8080, grpcPort, "127.0.0.1", "dn", map[string]uint32{"": 10}) + var volumes []storage.VolumeInfo + for vid := 1; vid <= 2; vid++ { + volumes = append(volumes, storage.VolumeInfo{ + Id: needle.VolumeId(vid), + Size: 1 << 20, + Collection: "c", + ModifiedAtSecond: time.Now().Unix(), + Version: needle.GetCurrentVersion(), + ReplicaPlacement: &super_block.ReplicaPlacement{}, + Ttl: needle.EMPTY_TTL, + }) + } + dn.UpdateVolumes(volumes) + for _, v := range volumes { + topo.RegisterVolumeLayout(v, dn) + } + committed := func() []uint32 { + fake.mu.Lock() + defer fake.mu.Unlock() + return append([]uint32(nil), fake.committed...) + } + + // the first request holds the guard for as long as its compaction is gated + first := make(chan error, 1) + go func() { + first <- topo.Vacuum(dialOption, 0.3, 1, 0, 1, "c", 0, false, 0) + }() + select { + case <-fake.entered: + case <-time.After(15 * time.Second): + t.Fatal("the first vacuum never reached the compact phase") + } + + if err := topo.Vacuum(dialOption, 0.3, 1, 0, 2, "c", 0, false, 0); !errors.Is(err, ErrVacuumRunning) { + t.Fatalf("vacuum of volume 2 while volume 1 is compacting: err = %v, want ErrVacuumRunning", err) + } + if got := committed(); len(got) != 0 { + t.Fatalf("committed volumes = %v, want none while the first vacuum is gated", got) + } + + close(fake.compactGate) + select { + case err := <-first: + if err != nil { + t.Fatalf("the first vacuum: %v", err) + } + case <-time.After(15 * time.Second): + t.Fatal("the first vacuum did not finish after the gate opened") + } + + // with the guard free again, the same request runs + if err := topo.Vacuum(dialOption, 0.3, 1, 0, 2, "c", 0, false, 0); err != nil { + t.Fatalf("vacuum of volume 2 after the first one finished: %v", err) + } + if got := committed(); len(got) != 2 || got[0] != 1 || got[1] != 2 { + t.Fatalf("committed volumes = %v, want [1 2]", got) + } +} + // A replica tiered to remote storage is never compacted by a vacuum, not even // on an explicit request with a zero threshold: its .dat is the remote object. // volume.tier.compact is the way to reclaim its garbage.