diff --git a/go.mod b/go.mod index ae6aa6f979..2a8b343b83 100644 --- a/go.mod +++ b/go.mod @@ -495,7 +495,7 @@ require ( golang.org/x/term v0.47.0 golang.org/x/time v0.16.0 google.golang.org/genproto/googleapis/api v0.0.0-20261012220534-4341edf3f941 // indirect - google.golang.org/genproto/googleapis/rpc v0.0.0-20261019180522-ce908d29299b // indirect + google.golang.org/genproto/googleapis/rpc v0.0.0-20261019180522-ce908d29299b gopkg.in/natefinch/lumberjack.v2 v2.2.1 // indirect gopkg.in/validator.v2 v2.0.1 // indirect gopkg.in/yaml.v2 v2.4.0 // indirect diff --git a/weed/operation/assign_file_id_retry_test.go b/weed/operation/assign_file_id_retry_test.go index ab9263dc02..019027c234 100644 --- a/weed/operation/assign_file_id_retry_test.go +++ b/weed/operation/assign_file_id_retry_test.go @@ -11,10 +11,12 @@ import ( "github.com/seaweedfs/seaweedfs/weed/pb" "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "google.golang.org/genproto/googleapis/rpc/errdetails" "google.golang.org/grpc" "google.golang.org/grpc/codes" "google.golang.org/grpc/credentials/insecure" "google.golang.org/grpc/status" + "google.golang.org/protobuf/types/known/durationpb" ) // fakeAssignServer returns Unavailable for the first N calls, then succeeds. @@ -106,3 +108,42 @@ func TestAssignStopsOnContextCancel(t *testing.T) { t.Errorf("expected context.DeadlineExceeded, got: %v", err) } } + +// warmingAssignServer refuses Assign until warmEnd, like a master in warm-up, +// telling the client the remaining warm-up. +type warmingAssignServer struct { + master_pb.UnimplementedSeaweedServer + warmEnd time.Time + callCount atomic.Int32 +} + +func (s *warmingAssignServer) Assign(_ context.Context, _ *master_pb.AssignRequest) (*master_pb.AssignResponse, error) { + s.callCount.Add(1) + if remaining := time.Until(s.warmEnd); remaining > 0 { + st, _ := status.New(codes.Unavailable, "master is warming up").WithDetails(&errdetails.RetryInfo{RetryDelay: durationpb.New(remaining)}) + return nil, st.Err() + } + return &master_pb.AssignResponse{Fid: "1,abc", Count: 1, Location: &master_pb.Location{Url: "127.0.0.1:8080", PublicUrl: "127.0.0.1:8080"}}, nil +} + +func TestAssignRetryLandsAtWarmupEndWithHint(t *testing.T) { + const warmup = 2600 * time.Millisecond + srv := &warmingAssignServer{} + addr := startFakeServer(t, srv) + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + start := time.Now() + srv.warmEnd = start.Add(warmup) + if _, err := Assign(ctx, func(_ context.Context) pb.ServerAddress { return addr }, + grpc.WithTransportCredentials(insecure.NewCredentials()), &VolumeAssignRequest{Count: 1}); err != nil { + t.Fatalf("Assign: %v", err) + } + // The hint lands the retry just after warm-up ends (2.6s), not at the + // next backoff step (1s + 1.5s + 2.25s = 4.75s). + if elapsed := time.Since(start); elapsed > 4500*time.Millisecond { + t.Fatalf("assign took %v over %d calls, want about 2.75s (warm-up end)", elapsed, srv.callCount.Load()) + } + if calls := srv.callCount.Load(); calls < 2 { + t.Fatalf("got %d calls, want a refused call during warm-up before the retry", calls) + } +} diff --git a/weed/server/master_grpc_server_assign.go b/weed/server/master_grpc_server_assign.go index 1fbe8fd843..4339d7732b 100644 --- a/weed/server/master_grpc_server_assign.go +++ b/weed/server/master_grpc_server_assign.go @@ -18,8 +18,10 @@ import ( "github.com/seaweedfs/seaweedfs/weed/storage/super_block" "github.com/seaweedfs/seaweedfs/weed/storage/types" "github.com/seaweedfs/seaweedfs/weed/topology" + "google.golang.org/genproto/googleapis/rpc/errdetails" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" + "google.golang.org/protobuf/types/known/durationpb" ) func (ms *MasterServer) StreamAssign(server master_pb.Seaweed_StreamAssignServer) error { @@ -71,7 +73,7 @@ func (ms *MasterServer) Assign(ctx context.Context, req *master_pb.AssignRequest } if ms.Topo.IsWarmingUp() { - return nil, status.Errorf(codes.Unavailable, "master is warming up, topology is still loading") + return nil, warmingUpError(ms.Topo.RemainingWarmupDuration()) } diskType := types.ToDiskType(req.DiskType) @@ -330,3 +332,17 @@ func (w *unservedLayoutWarning) Do(optionKey string, lastErr error) { w.last[optionKey] = now glog.Warningf("assign requests for %s will keep failing until a volume server registers that disk layout or clients change their assignment disk type: %v", optionKey, lastErr) } + +// warmingUpError carries the remaining warm-up as a RetryInfo detail, the gRPC +// counterpart of the HTTP handlers' Retry-After. +func warmingUpError(remaining time.Duration) error { + st := status.New(codes.Unavailable, "master is warming up, topology is still loading") + if remaining <= 0 { + return st.Err() + } + withHint, err := st.WithDetails(&errdetails.RetryInfo{RetryDelay: durationpb.New(remaining)}) + if err != nil { + return st.Err() + } + return withHint.Err() +} diff --git a/weed/server/master_grpc_server_assign_warmup_test.go b/weed/server/master_grpc_server_assign_warmup_test.go new file mode 100644 index 0000000000..d086bf2bcb --- /dev/null +++ b/weed/server/master_grpc_server_assign_warmup_test.go @@ -0,0 +1,49 @@ +package weed_server + +import ( + "context" + "testing" + + "github.com/seaweedfs/seaweedfs/weed/pb/master_pb" + "google.golang.org/genproto/googleapis/rpc/errdetails" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" +) + +// assertWarmupHint calls refuse during warm-up and checks it is refused with +// Unavailable and a RetryInfo delay within the remaining warm-up measured +// around the call. +func assertWarmupHint(t *testing.T, ms *MasterServer, refuse func() error) { + t.Helper() + before := ms.Topo.RemainingWarmupDuration() + err := refuse() + after := ms.Topo.RemainingWarmupDuration() + st, _ := status.FromError(err) + var hint *errdetails.RetryInfo + for _, d := range st.Details() { + if ri, ok := d.(*errdetails.RetryInfo); ok { + hint = ri + } + } + if st.Code() != codes.Unavailable || hint == nil { + t.Fatalf("refused with %v and details %v, want Unavailable with RetryInfo", err, st.Details()) + } + if got := hint.GetRetryDelay().AsDuration(); got <= 0 || got > before || got < after { + t.Fatalf("RetryInfo delay = %v, want within [%v, %v]", got, after, before) + } +} + +func TestAssignDuringWarmupCarriesRemainingWarmup(t *testing.T) { + ms := newWarmingUpMaster(t) + assertWarmupHint(t, ms, func() error { + _, err := ms.Assign(context.Background(), &master_pb.AssignRequest{Count: 1}) + return err + }) +} + +func TestWarmingUpErrorWithoutRemainingHasNoHint(t *testing.T) { + st, _ := status.FromError(warmingUpError(0)) + if st.Code() != codes.Unavailable || len(st.Details()) != 0 { + t.Fatalf("got %v with details %v, want plain Unavailable", st.Code(), st.Details()) + } +} diff --git a/weed/server/master_grpc_server_volume.go b/weed/server/master_grpc_server_volume.go index b38cc52450..96b60182f9 100644 --- a/weed/server/master_grpc_server_volume.go +++ b/weed/server/master_grpc_server_volume.go @@ -228,7 +228,7 @@ func (ms *MasterServer) LookupVolume(ctx context.Context, req *master_pb.LookupV // missing volume as gone. if notFoundCount > 0 && ms.Topo.IsLeader() && ms.Topo.IsWarmingUp() { glog.V(0).Infof("lookup volume warming up: topology is still loading (%d of %d not found)", notFoundCount, len(req.VolumeOrFileIds)) - return nil, status.Errorf(codes.Unavailable, "master is warming up, topology is still loading") + return nil, warmingUpError(ms.Topo.RemainingWarmupDuration()) } return resp, nil diff --git a/weed/server/master_grpc_server_volume_warmup_test.go b/weed/server/master_grpc_server_volume_warmup_test.go index b62d4a32d7..4d0e8ca2ee 100644 --- a/weed/server/master_grpc_server_volume_warmup_test.go +++ b/weed/server/master_grpc_server_volume_warmup_test.go @@ -61,3 +61,11 @@ func TestLookupVolumeAfterWarmupReportsNotFoundPerVolume(t *testing.T) { t.Fatalf("absent volume answer = %+v", resp.VolumeIdLocations[1]) } } + +func TestLookupVolumeDuringWarmupCarriesRemainingWarmup(t *testing.T) { + ms := newWarmingUpMaster(t) + assertWarmupHint(t, ms, func() error { + _, err := ms.LookupVolume(context.Background(), &master_pb.LookupVolumeRequest{VolumeOrFileIds: []string{"7", "8"}}) + return err + }) +} diff --git a/weed/util/retry.go b/weed/util/retry.go index 6f77be9115..dfb8c427ec 100644 --- a/weed/util/retry.go +++ b/weed/util/retry.go @@ -4,12 +4,14 @@ import ( "context" "errors" "io" + "math/rand/v2" "net" "strings" "syscall" "time" "github.com/seaweedfs/seaweedfs/weed/glog" + "google.golang.org/genproto/googleapis/rpc/errdetails" "google.golang.org/grpc/codes" "google.golang.org/grpc/status" ) @@ -246,11 +248,12 @@ func RetryWithBackoff(ctx context.Context, name string, maxDuration time.Duratio if sleepTime > maxWaitTime { sleepTime = maxWaitTime } + sleepTime = shortenStep(sleepTime, err, retryHintSlack()) if sleepTime > remaining { sleepTime = remaining } glog.V(1).Infof("retry %s: retrying in %v: %v", name, sleepTime, err) - timer := time.NewTimer(sleepTime) + timer := retryBackoffTimer(sleepTime) select { case <-ctx.Done(): if !timer.Stop() { @@ -266,6 +269,40 @@ func RetryWithBackoff(ctx context.Context, name string, maxDuration time.Duratio } } +var retryBackoffTimer = time.NewTimer + +// retryHintSlack lands a hinted retry just after the server is ready, jittered +// so clients refused together do not retry together. +func retryHintSlack() time.Duration { + return 50*time.Millisecond + rand.N(250*time.Millisecond+1) +} + +// shortenStep returns hint plus slack when that is shorter than step. +// hint < step-slack, not hint+slack < step, so a huge hint cannot overflow. +func shortenStep(step time.Duration, err error, slack time.Duration) time.Duration { + if hint, ok := serverRetryDelay(err); ok && hint < step-slack { + return hint + slack + } + return step +} + +// serverRetryDelay returns a valid, positive gRPC RetryInfo delay; anything +// else would let a misbehaving server turn the backoff into tight polling. +func serverRetryDelay(err error) (time.Duration, bool) { + st, ok := ServerStatus(err) + if !ok { + return 0, false + } + for _, d := range st.Details() { + if ri, ok := d.(*errdetails.RetryInfo); ok && ri.GetRetryDelay().CheckValid() == nil { + if delay := ri.GetRetryDelay().AsDuration(); delay > 0 { + return delay, true + } + } + } + return 0, false +} + // Nvl return the first non-empty string func Nvl(values ...string) string { for _, s := range values { diff --git a/weed/util/retry_hint_test.go b/weed/util/retry_hint_test.go new file mode 100644 index 0000000000..460b0d9463 --- /dev/null +++ b/weed/util/retry_hint_test.go @@ -0,0 +1,95 @@ +package util + +import ( + "context" + "errors" + "fmt" + "math" + "testing" + "time" + + "google.golang.org/genproto/googleapis/rpc/errdetails" + "google.golang.org/grpc/codes" + "google.golang.org/grpc/status" + "google.golang.org/protobuf/types/known/durationpb" +) + +func unavailableWithDelay(t *testing.T, d *durationpb.Duration) error { + t.Helper() + st, err := status.New(codes.Unavailable, "warming up").WithDetails(&errdetails.RetryInfo{RetryDelay: d}) + if err != nil { + t.Fatalf("WithDetails: %v", err) + } + return st.Err() +} + +// Every guard on a server hint, without sleeping: a hint only ever shortens +// the backoff step, and only a valid, positive one counts. +func TestShortenStep(t *testing.T) { + const step, slack = time.Second, 50 * time.Millisecond + hinted := func(d time.Duration) error { return unavailableWithDelay(t, durationpb.New(d)) } + for _, c := range []struct { + name string + err error + want time.Duration + }{ + {"valid shorter hint", hinted(200 * time.Millisecond), 250 * time.Millisecond}, + {"wrapped status keeps its hint", fmt.Errorf("assign: %w", hinted(200*time.Millisecond)), 250 * time.Millisecond}, + {"zero hint", hinted(0), step}, + {"negative hint", hinted(-200 * time.Millisecond), step}, + {"hint longer than the step", hinted(10 * time.Second), step}, + {"huge hint does not overflow", hinted(time.Duration(math.MaxInt64)), step}, + {"malformed duration", unavailableWithDelay(t, &durationpb.Duration{Seconds: 1, Nanos: -999999999}), step}, + {"no hint", status.Error(codes.Unavailable, "warming up"), step}, + {"not a status error", errors.New("plain"), step}, + } { + if got := shortenStep(step, c.err, slack); got != c.want { + t.Errorf("%s: step %v, want %v", c.name, got, c.want) + } + } +} + +// The slack is jittered within [50ms, 300ms] so refused clients spread out. +func TestRetryHintSlackIsJitteredWithinBounds(t *testing.T) { + lo, hi := time.Hour, time.Duration(0) + for i := 0; i < 2000; i++ { + s := retryHintSlack() + if s < 50*time.Millisecond || s > 300*time.Millisecond { + t.Fatalf("slack %v outside [50ms, 300ms]", s) + } + lo, hi = min(lo, s), max(hi, s) + } + if hi-lo < 100*time.Millisecond { + t.Fatalf("slack barely varies: [%v, %v]", lo, hi) + } +} + +// The loop waits hint + slack, not the 1s step, after a refusal with a 100ms hint. +func TestRetryWithBackoffHonoursShorterServerHint(t *testing.T) { + var waits []time.Duration + defer func(orig func(time.Duration) *time.Timer) { retryBackoffTimer = orig }(retryBackoffTimer) + retryBackoffTimer = func(d time.Duration) *time.Timer { + waits = append(waits, d) + return time.NewTimer(d) + } + start := time.Now() + var calls int + err := RetryWithBackoff(context.Background(), "hint-test", 30*time.Second, + func(err error) bool { return status.Code(err) == codes.Unavailable }, + func() error { + calls++ + if calls == 1 { + return unavailableWithDelay(t, durationpb.New(100*time.Millisecond)) + } + return nil + }) + if err != nil || calls != 2 { + t.Fatalf("RetryWithBackoff = %v after %d calls, want success after 2", err, calls) + } + if elapsed := time.Since(start); elapsed < 150*time.Millisecond { + t.Fatalf("returned after %v, want the retry to wait out its timer", elapsed) + } + if len(waits) != 1 || waits[0] < 150*time.Millisecond || waits[0] > 400*time.Millisecond { + t.Fatalf("waits %v, want one wait of hint + slack (150-400ms), not the 1s step", waits) + } +}