filer: replicate RECOMPUTE_LATEST pointer updates to peers (#9840)
applyRecomputeLatest wrote the .versions latest-version pointer and the demoted prior version's stamp through UpdateEntry without a following NotifyUpdateEvent, so neither change entered the metadata log. Across filers the pointer then lived only on whichever filer ran the mutation, and ListObjects served by any other filer dropped those objects from a versioned bucket. Emit the events the way PATCH_EXTENDED already does, keeping a pre-update image for the notification diff.
This commit is contained in:
Chris Lu
GitHub
parent
6e16994615
commit
9ede92a7cc
@ -425,7 +425,7 @@ func (fs *FilerServer) applyObjectMutation(ctx context.Context, m *filer_pb.Obje
|
||||
return nil
|
||||
|
||||
case filer_pb.ObjectMutation_RECOMPUTE_LATEST:
|
||||
return fs.applyRecomputeLatest(ctx, m)
|
||||
return fs.applyRecomputeLatest(ctx, m, fromOtherCluster, signatures)
|
||||
|
||||
default:
|
||||
return fmt.Errorf("unknown mutation type %v", m.Type)
|
||||
@ -439,7 +439,7 @@ func (fs *FilerServer) applyObjectMutation(ctx context.Context, m *filer_pb.Obje
|
||||
// name_to_key. When the scanned directory is empty the pointer keys are cleared.
|
||||
// The caller, which knows the versioning scheme, supplies the direction and the
|
||||
// key mappings. A missing pointer entry is a no-op (idempotent on replay).
|
||||
func (fs *FilerServer) applyRecomputeLatest(ctx context.Context, m *filer_pb.ObjectMutation) error {
|
||||
func (fs *FilerServer) applyRecomputeLatest(ctx context.Context, m *filer_pb.ObjectMutation, fromOtherCluster bool, signatures []int32) error {
|
||||
rc := m.Recompute
|
||||
if rc == nil {
|
||||
return fmt.Errorf("RECOMPUTE_LATEST requires recompute parameters")
|
||||
@ -452,6 +452,15 @@ func (fs *FilerServer) applyRecomputeLatest(ctx context.Context, m *filer_pb.Obj
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
// Capture the pre-update image so the metadata notification carries a correct
|
||||
// diff; pointer.Extended is mutated in place below.
|
||||
oldPointer := pointer.ShallowClone()
|
||||
oldPointer.Extended = make(map[string][]byte, len(pointer.Extended))
|
||||
for k, v := range pointer.Extended {
|
||||
oldPointer.Extended[k] = v
|
||||
}
|
||||
|
||||
if pointer.Extended == nil {
|
||||
pointer.Extended = make(map[string][]byte)
|
||||
}
|
||||
@ -512,9 +521,13 @@ func (fs *FilerServer) applyRecomputeLatest(ctx context.Context, m *filer_pb.Obj
|
||||
}
|
||||
}
|
||||
|
||||
if err := fs.filer.UpdateEntry(ctx, pointer, pointer); err != nil {
|
||||
if err := fs.filer.UpdateEntry(ctx, oldPointer, pointer); err != nil {
|
||||
return err
|
||||
}
|
||||
// Replicate the recomputed pointer to peer filers and subscribers. Without
|
||||
// this the latest-version pointer stays in this filer's store only, so other
|
||||
// filers never learn the current version and ListObjects undercounts.
|
||||
fs.filer.NotifyUpdateEvent(ctx, oldPointer, pointer, false, fromOtherCluster, signatures)
|
||||
|
||||
// Stamp the displaced prior child (e.g. NoncurrentSinceNs for lifecycle).
|
||||
newName := ""
|
||||
@ -529,11 +542,20 @@ func (fs *FilerServer) applyRecomputeLatest(ctx context.Context, m *filer_pb.Obj
|
||||
if perr != nil {
|
||||
return perr
|
||||
}
|
||||
oldPrior := priorEntry.ShallowClone()
|
||||
oldPrior.Extended = make(map[string][]byte, len(priorEntry.Extended))
|
||||
for k, v := range priorEntry.Extended {
|
||||
oldPrior.Extended[k] = v
|
||||
}
|
||||
if priorEntry.Extended == nil {
|
||||
priorEntry.Extended = make(map[string][]byte)
|
||||
}
|
||||
priorEntry.Extended[rc.DemoteKey] = rc.DemoteValue
|
||||
return fs.filer.UpdateEntry(ctx, priorEntry, priorEntry)
|
||||
if err := fs.filer.UpdateEntry(ctx, oldPrior, priorEntry); err != nil {
|
||||
return err
|
||||
}
|
||||
fs.filer.NotifyUpdateEvent(ctx, oldPrior, priorEntry, false, fromOtherCluster, signatures)
|
||||
return nil
|
||||
}
|
||||
|
||||
return nil
|
||||
|
||||
140
weed/server/filer_grpc_server_recompute_test.go
Normal file
140
weed/server/filer_grpc_server_recompute_test.go
Normal file
@ -0,0 +1,140 @@
|
||||
package weed_server
|
||||
|
||||
import (
|
||||
"context"
|
||||
"testing"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/filer_pb"
|
||||
)
|
||||
|
||||
// RECOMPUTE_LATEST must emit a metadata notification for the pointer it writes.
|
||||
// Without it the recomputed latest-version pointer lives only in the filer that
|
||||
// processed the mutation, so peer filers never learn the current version and
|
||||
// their ListObjects undercount versioned buckets.
|
||||
func TestRecomputeLatestEmitsPointerUpdateEvent(t *testing.T) {
|
||||
store := newRenameTestStore()
|
||||
store.entries["/buckets/b/obj.versions"] = newDirectoryEntry("/buckets/b/obj.versions", 10)
|
||||
version := newFileEntry("/buckets/b/obj.versions/v_123", 11)
|
||||
version.Extended = map[string][]byte{"vid": []byte("123")}
|
||||
store.entries["/buckets/b/obj.versions/v_123"] = version
|
||||
|
||||
queue := &captureQueue{}
|
||||
swapNotificationQueue(t, queue)
|
||||
|
||||
server := &FilerServer{filer: newRenameTestFiler(store)}
|
||||
m := &filer_pb.ObjectMutation{
|
||||
Type: filer_pb.ObjectMutation_RECOMPUTE_LATEST,
|
||||
Directory: "/buckets/b",
|
||||
Name: "obj.versions",
|
||||
Recompute: &filer_pb.Recompute{
|
||||
ScanDir: "/buckets/b/obj.versions",
|
||||
Descending: true,
|
||||
NameToKey: "latest-file",
|
||||
CopyExtended: map[string]string{"latest-vid": "vid"},
|
||||
},
|
||||
}
|
||||
if err := server.applyObjectMutation(context.Background(), m, false, nil); err != nil {
|
||||
t.Fatalf("applyObjectMutation: %v", err)
|
||||
}
|
||||
|
||||
// Pointer is persisted locally.
|
||||
ptr, err := store.FindEntry(context.Background(), "/buckets/b/obj.versions")
|
||||
if err != nil {
|
||||
t.Fatalf("find pointer: %v", err)
|
||||
}
|
||||
if got := string(ptr.Extended["latest-vid"]); got != "123" {
|
||||
t.Fatalf("pointer latest-vid = %q, want 123", got)
|
||||
}
|
||||
|
||||
// ...and it is announced so peers replicate it.
|
||||
events := queue.snapshot()
|
||||
if len(events) != 1 {
|
||||
t.Fatalf("event count = %d, want 1", len(events))
|
||||
}
|
||||
e := events[0]
|
||||
if e.notification.NewEntry == nil || e.notification.NewEntry.Name != "obj.versions" {
|
||||
t.Fatalf("event new entry = %+v, want obj.versions", e.notification.NewEntry)
|
||||
}
|
||||
if got := string(e.notification.NewEntry.Extended["latest-vid"]); got != "123" {
|
||||
t.Fatalf("event latest-vid = %q, want 123", got)
|
||||
}
|
||||
if e.notification.OldEntry != nil && len(e.notification.OldEntry.Extended["latest-vid"]) != 0 {
|
||||
t.Fatalf("event old entry already had pointer: %+v", e.notification.OldEntry)
|
||||
}
|
||||
}
|
||||
|
||||
// A pointer flip that demotes the prior latest must announce both the pointer
|
||||
// update and the demote stamp, so a peer's view of both entries stays correct.
|
||||
func TestRecomputeLatestDemoteEmitsEvent(t *testing.T) {
|
||||
store := newRenameTestStore()
|
||||
pointer := newDirectoryEntry("/buckets/b/obj.versions", 10)
|
||||
pointer.Extended = map[string][]byte{"latest-file": []byte("v_100"), "latest-vid": []byte("100")}
|
||||
store.entries["/buckets/b/obj.versions"] = pointer
|
||||
|
||||
prior := newFileEntry("/buckets/b/obj.versions/v_100", 11)
|
||||
prior.Extended = map[string][]byte{"vid": []byte("100")}
|
||||
store.entries["/buckets/b/obj.versions/v_100"] = prior
|
||||
|
||||
latest := newFileEntry("/buckets/b/obj.versions/v_200", 12)
|
||||
latest.Extended = map[string][]byte{"vid": []byte("200")}
|
||||
store.entries["/buckets/b/obj.versions/v_200"] = latest
|
||||
|
||||
queue := &captureQueue{}
|
||||
swapNotificationQueue(t, queue)
|
||||
|
||||
server := &FilerServer{filer: newRenameTestFiler(store)}
|
||||
m := &filer_pb.ObjectMutation{
|
||||
Type: filer_pb.ObjectMutation_RECOMPUTE_LATEST,
|
||||
Directory: "/buckets/b",
|
||||
Name: "obj.versions",
|
||||
Recompute: &filer_pb.Recompute{
|
||||
ScanDir: "/buckets/b/obj.versions",
|
||||
Descending: true,
|
||||
NameToKey: "latest-file",
|
||||
CopyExtended: map[string]string{"latest-vid": "vid"},
|
||||
DemoteKey: "noncurrent-since",
|
||||
DemoteValue: []byte("999"),
|
||||
},
|
||||
}
|
||||
if err := server.applyObjectMutation(context.Background(), m, false, nil); err != nil {
|
||||
t.Fatalf("applyObjectMutation: %v", err)
|
||||
}
|
||||
|
||||
gotPtr, err := store.FindEntry(context.Background(), "/buckets/b/obj.versions")
|
||||
if err != nil {
|
||||
t.Fatalf("find pointer: %v", err)
|
||||
}
|
||||
if got := string(gotPtr.Extended["latest-file"]); got != "v_200" {
|
||||
t.Fatalf("pointer latest-file = %q, want v_200", got)
|
||||
}
|
||||
gotPrior, err := store.FindEntry(context.Background(), "/buckets/b/obj.versions/v_100")
|
||||
if err != nil {
|
||||
t.Fatalf("find prior: %v", err)
|
||||
}
|
||||
if got := string(gotPrior.Extended["noncurrent-since"]); got != "999" {
|
||||
t.Fatalf("prior noncurrent-since = %q, want 999", got)
|
||||
}
|
||||
|
||||
var sawPointer, sawDemote bool
|
||||
for _, e := range queue.snapshot() {
|
||||
if e.notification.NewEntry == nil {
|
||||
continue
|
||||
}
|
||||
switch e.notification.NewEntry.Name {
|
||||
case "obj.versions":
|
||||
if string(e.notification.NewEntry.Extended["latest-file"]) == "v_200" {
|
||||
sawPointer = true
|
||||
}
|
||||
case "v_100":
|
||||
if string(e.notification.NewEntry.Extended["noncurrent-since"]) == "999" {
|
||||
sawDemote = true
|
||||
}
|
||||
}
|
||||
}
|
||||
if !sawPointer {
|
||||
t.Fatalf("missing pointer-flip event: %+v", queue.snapshot())
|
||||
}
|
||||
if !sawDemote {
|
||||
t.Fatalf("missing demote event: %+v", queue.snapshot())
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue
Block a user
