From 84aff673c2638575728f1460bb6fa41dc95759e8 Mon Sep 17 00:00:00 2001 From: Chepurnov Simon <195125934+CHEPURNOV-S-S@users.noreply.github.com> Date: Sun, 11 Oct 2026 09:28:23 +0000 Subject: [PATCH] feat: prefer same-host worker and local EC replica Admin executor pick ranked idle workers only, so a job whose volume lived on full-03 could still run on full-01 and pull the whole .dat across the LAN. Rank workers whose Hello host matches any replica first (fallback to others if the local worker is full). Detection named Sources[0] / the lowest server id; execute then SyncAndSelectBestReplica and could switch the copy source to another rack. After sync, if a replica on this machine has a complete file count, copy from that URL so affinity and pull stay on one host. --- weed/admin/plugin/host_affinity.go | 66 ++++++++++++++++ weed/admin/plugin/plugin.go | 2 +- weed/admin/plugin/plugin_scheduler.go | 5 +- weed/admin/plugin/plugin_scheduler_test.go | 10 +-- weed/admin/plugin/registry.go | 28 +++++-- weed/admin/plugin/registry_test.go | 67 ++++++++++++++++ weed/util/net_host.go | 79 +++++++++++++++++++ weed/util/net_host_test.go | 24 ++++++ weed/worker/tasks/erasure_coding/ec_task.go | 44 +++++++++++ .../tasks/erasure_coding/plugin_handler.go | 14 +++- 10 files changed, 324 insertions(+), 15 deletions(-) create mode 100644 weed/admin/plugin/host_affinity.go create mode 100644 weed/util/net_host.go create mode 100644 weed/util/net_host_test.go diff --git a/weed/admin/plugin/host_affinity.go b/weed/admin/plugin/host_affinity.go new file mode 100644 index 0000000000..ddff60ece0 --- /dev/null +++ b/weed/admin/plugin/host_affinity.go @@ -0,0 +1,66 @@ +package plugin + +import ( + "strings" + + "github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb" + "github.com/seaweedfs/seaweedfs/weed/pb/worker_pb" + pluginworker "github.com/seaweedfs/seaweedfs/weed/plugin/worker" + "github.com/seaweedfs/seaweedfs/weed/util" + + "google.golang.org/protobuf/proto" +) + +// JobReplicaHosts are the volume-replica hosts a job will read from. +// Used so executor pick prefers a worker on one of those machines. +// Sources with ShardIds are EC-cleanup rows, not encode replicas. +func JobReplicaHosts(job *plugin_pb.JobSpec) []string { + if job == nil { + return nil + } + var addrs []string + addrs = append(addrs, pluginworker.ReadStringListConfig(job.Parameters, "replica_servers")...) + if src := strings.TrimSpace(pluginworker.ReadStringConfig(job.Parameters, "source_server", "")); src != "" { + addrs = append(addrs, src) + } + if payload := pluginworker.ReadBytesConfig(job.Parameters, "task_params_pb"); len(payload) > 0 { + params := &worker_pb.TaskParams{} + if err := proto.Unmarshal(payload, params); err == nil { + for _, s := range params.GetSources() { + if s == nil || s.Node == "" || len(s.ShardIds) > 0 { + continue + } + addrs = append(addrs, s.Node) + } + } + } + return util.HostKeysFromAddresses(addrs) +} + +// partitionByPreferredHost splits candidates into (host matches, others). +// No preferred hosts, or no matches, returns (nil, all) so callers keep +// the existing slots+round-robin path. +func partitionByPreferredHost(candidates []*WorkerSession, preferredHosts []string) (pref, rest []*WorkerSession) { + want := util.HostKeysFromAddresses(preferredHosts) + if len(want) == 0 || len(candidates) == 0 { + return nil, candidates + } + set := make(map[string]struct{}, len(want)) + for _, h := range want { + set[h] = struct{}{} + } + for _, c := range candidates { + if c == nil { + continue + } + if _, ok := set[util.HostKey(c.Address)]; ok { + pref = append(pref, c) + continue + } + rest = append(rest, c) + } + if len(pref) == 0 { + return nil, candidates + } + return pref, rest +} diff --git a/weed/admin/plugin/plugin.go b/weed/admin/plugin/plugin.go index e94e0fc798..1dcfe52f4c 100644 --- a/weed/admin/plugin/plugin.go +++ b/weed/admin/plugin/plugin.go @@ -609,7 +609,7 @@ func (r *Plugin) ExecuteJob( releaseGate := r.acquireDetectionExecutionGate(jobType, true) defer releaseGate() - executor, err := r.registry.PickExecutor(jobType) + executor, err := r.registry.PickExecutor(jobType, JobReplicaHosts(job)...) if err != nil { return nil, err } diff --git a/weed/admin/plugin/plugin_scheduler.go b/weed/admin/plugin/plugin_scheduler.go index 94a3aa3976..4d60ca9458 100644 --- a/weed/admin/plugin/plugin_scheduler.go +++ b/weed/admin/plugin/plugin_scheduler.go @@ -873,7 +873,7 @@ func (r *Plugin) dispatchScheduledProposals( continue jobLoop } - executor, release, reserveErr := r.reserveScheduledExecutor(ctx, jobType, policy) + executor, release, reserveErr := r.reserveScheduledExecutor(ctx, jobType, policy, JobReplicaHosts(job)) if reserveErr != nil { if ctx.Err() != nil { r.cancelQueuedJob(job, ctx.Err()) @@ -988,6 +988,7 @@ func (r *Plugin) reserveScheduledExecutor( ctx context.Context, jobType string, policy schedulerPolicy, + preferredHosts []string, ) (*WorkerSession, func(), error) { if ctx == nil { ctx = context.Background() @@ -1015,7 +1016,7 @@ func (r *Plugin) reserveScheduledExecutor( return nil, nil, fmt.Errorf("timed out waiting for executor capacity for %s", jobType) } - executors, err := r.registry.ListExecutors(jobType) + executors, err := r.registry.ListExecutors(jobType, preferredHosts...) if err != nil { if !waitForShutdownOrTimerWithContext(r.shutdownCh, ctx, policy.ExecutorReserveBackoff) { if ctx.Err() != nil { diff --git a/weed/admin/plugin/plugin_scheduler_test.go b/weed/admin/plugin/plugin_scheduler_test.go index 2e9459bb95..186b8b5a39 100644 --- a/weed/admin/plugin/plugin_scheduler_test.go +++ b/weed/admin/plugin/plugin_scheduler_test.go @@ -138,13 +138,13 @@ func TestReserveScheduledExecutorRespectsPerWorkerLimit(t *testing.T) { ExecutorReserveBackoff: time.Millisecond, } - executor1, release1, err := pluginSvc.reserveScheduledExecutor(context.Background(), "balance", policy) + executor1, release1, err := pluginSvc.reserveScheduledExecutor(context.Background(), "balance", policy, nil) if err != nil { t.Fatalf("reserve executor 1: %v", err) } defer release1() - executor2, release2, err := pluginSvc.reserveScheduledExecutor(context.Background(), "balance", policy) + executor2, release2, err := pluginSvc.reserveScheduledExecutor(context.Background(), "balance", policy, nil) if err != nil { t.Fatalf("reserve executor 2: %v", err) } @@ -266,7 +266,7 @@ func TestReserveScheduledExecutorTimesOutWhenNoExecutor(t *testing.T) { start := time.Now() pluginSvc.Shutdown() - _, _, err = pluginSvc.reserveScheduledExecutor(context.Background(), "missing-job-type", policy) + _, _, err = pluginSvc.reserveScheduledExecutor(context.Background(), "missing-job-type", policy, nil) if err == nil { t.Fatalf("expected reservation shutdown error") } @@ -297,7 +297,7 @@ func TestReserveScheduledExecutorWaitsForWorkerCapacity(t *testing.T) { ExecutorReserveBackoff: 5 * time.Millisecond, } - _, release1, err := pluginSvc.reserveScheduledExecutor(context.Background(), "balance", policy) + _, release1, err := pluginSvc.reserveScheduledExecutor(context.Background(), "balance", policy, nil) if err != nil { t.Fatalf("reserve executor 1: %v", err) } @@ -308,7 +308,7 @@ func TestReserveScheduledExecutorWaitsForWorkerCapacity(t *testing.T) { } secondReserveCh := make(chan reserveResult, 1) go func() { - _, release2, reserveErr := pluginSvc.reserveScheduledExecutor(context.Background(), "balance", policy) + _, release2, reserveErr := pluginSvc.reserveScheduledExecutor(context.Background(), "balance", policy, nil) if release2 != nil { release2() } diff --git a/weed/admin/plugin/registry.go b/weed/admin/plugin/registry.go index 46d51e79d9..2201e6a9c5 100644 --- a/weed/admin/plugin/registry.go +++ b/weed/admin/plugin/registry.go @@ -277,14 +277,24 @@ func (r *Registry) PickDetector(jobType string) (*WorkerSession, error) { } // PickExecutor picks one executor worker for a job type. -func (r *Registry) PickExecutor(jobType string) (*WorkerSession, error) { - return r.pickByKind(jobType, false) +// preferredHosts, when set, ranks workers whose Hello Address host matches a +// replica/source host ahead of others (same-machine EC pull). Empty means the +// previous slots+round-robin order. +func (r *Registry) PickExecutor(jobType string, preferredHosts ...string) (*WorkerSession, error) { + list, err := r.ListExecutors(jobType, preferredHosts...) + if err != nil { + return nil, err + } + return list[0], nil } // ListExecutors returns sorted executor candidates for one job type. // Ordering is by most available execution slots, then lexical worker ID. // The top tie group is rotated round-robin to prevent sticky assignment. -func (r *Registry) ListExecutors(jobType string) ([]*WorkerSession, error) { +// preferredHosts, when any worker matches, are listed first (then the rest); +// rotation applies only inside the preferred group so a local worker is not +// rotated away in favor of a remote one. +func (r *Registry) ListExecutors(jobType string, preferredHosts ...string) ([]*WorkerSession, error) { r.mu.Lock() defer r.mu.Unlock() @@ -293,8 +303,16 @@ func (r *Registry) ListExecutors(jobType string) ([]*WorkerSession, error) { return nil, fmt.Errorf("no executor worker available for job_type=%s", jobType) } - sortByKind(candidates, jobType, false) - r.rotateTopCandidatesLocked(candidates, jobType, false) + pref, rest := partitionByPreferredHost(candidates, preferredHosts) + if len(pref) > 0 { + sortByKind(pref, jobType, false) + r.rotateTopCandidatesLocked(pref, jobType, false) + sortByKind(rest, jobType, false) + candidates = append(pref, rest...) + } else { + sortByKind(candidates, jobType, false) + r.rotateTopCandidatesLocked(candidates, jobType, false) + } out := make([]*WorkerSession, 0, len(candidates)) for _, candidate := range candidates { diff --git a/weed/admin/plugin/registry_test.go b/weed/admin/plugin/registry_test.go index 1d331278cb..d023e3ec99 100644 --- a/weed/admin/plugin/registry_test.go +++ b/weed/admin/plugin/registry_test.go @@ -401,3 +401,70 @@ func TestRegistryHasCapableWorkerIgnoresNilCapability(t *testing.T) { t.Fatalf("expected HasCapableWorker to return false when capability has no detect or execute") } } + +func TestListExecutorsPrefersMatchingReplicaHost(t *testing.T) { + t.Parallel() + + r := NewRegistry() + r.UpsertFromHello(&plugin_pb.WorkerHello{ + WorkerId: "worker-remote", + Address: "10.87.0.4:33646", + Capabilities: []*plugin_pb.JobTypeCapability{ + {JobType: "erasure_coding", CanExecute: true, MaxExecutionConcurrency: 8}, + }, + }) + r.UpsertFromHello(&plugin_pb.WorkerHello{ + WorkerId: "worker-local", + Address: "10.87.0.3:33646", + Capabilities: []*plugin_pb.JobTypeCapability{ + {JobType: "erasure_coding", CanExecute: true, MaxExecutionConcurrency: 1}, + }, + }) + r.UpdateHeartbeat("worker-remote", &plugin_pb.WorkerHeartbeat{ + WorkerId: "worker-remote", ExecutionSlotsUsed: 0, ExecutionSlotsTotal: 8, + }) + r.UpdateHeartbeat("worker-local", &plugin_pb.WorkerHeartbeat{ + WorkerId: "worker-local", ExecutionSlotsUsed: 0, ExecutionSlotsTotal: 1, + }) + + // Without preference the emptier remote worker wins on slots. + plain, err := r.ListExecutors("erasure_coding") + if err != nil { + t.Fatalf("ListExecutors: %v", err) + } + if plain[0].WorkerID != "worker-remote" { + t.Fatalf("without preference want remote first, got %s", plain[0].WorkerID) + } + + // Replica on :8080 of the same host as worker-local. + pref, err := r.ListExecutors("erasure_coding", "10.87.0.3:8080") + if err != nil { + t.Fatalf("ListExecutors preferred: %v", err) + } + if pref[0].WorkerID != "worker-local" { + t.Fatalf("preferred host want worker-local first, got %s", pref[0].WorkerID) + } + if pref[1].WorkerID != "worker-remote" { + t.Fatalf("remote should still be fallback, got %s", pref[1].WorkerID) + } +} + +func TestJobReplicaHostsFromSourcesAndReplicaServers(t *testing.T) { + t.Parallel() + job := &plugin_pb.JobSpec{ + Parameters: map[string]*plugin_pb.ConfigValue{ + "source_server": { + Kind: &plugin_pb.ConfigValue_StringValue{StringValue: "10.87.0.1:8080"}, + }, + "replica_servers": { + Kind: &plugin_pb.ConfigValue_StringList{ + StringList: &plugin_pb.StringList{Values: []string{"10.87.0.1:8080", "10.87.0.2:8081"}}, + }, + }, + }, + } + got := JobReplicaHosts(job) + if len(got) != 2 { + t.Fatalf("got %v", got) + } +} diff --git a/weed/util/net_host.go b/weed/util/net_host.go new file mode 100644 index 0000000000..aa6fd385ca --- /dev/null +++ b/weed/util/net_host.go @@ -0,0 +1,79 @@ +package util + +import ( + "net" + "strings" +) + +// HostKey is the hostname/IP of an "host:port" address (or the string itself +// when it has no port). Compared case-insensitively so worker Hello +// "10.87.0.3:33646" matches a volume replica on "10.87.0.3:8080". +func HostKey(addr string) string { + addr = strings.TrimSpace(addr) + if addr == "" { + return "" + } + host, _, err := net.SplitHostPort(addr) + if err != nil { + return strings.ToLower(addr) + } + return strings.ToLower(host) +} + +// HostKeysFromAddresses returns unique HostKey values, skipping empties. +func HostKeysFromAddresses(addrs []string) []string { + seen := make(map[string]struct{}, len(addrs)) + var out []string + for _, a := range addrs { + h := HostKey(a) + if h == "" { + continue + } + if _, ok := seen[h]; ok { + continue + } + seen[h] = struct{}{} + out = append(out, h) + } + return out +} + +// AddressIsLocal reports whether addr's host is this process's machine +// (loopback or a configured interface address). Used so an EC worker prefers +// a replica on the same box after replica sync. +func AddressIsLocal(addr string) bool { + host := HostKey(addr) + if host == "" { + return false + } + if host == "localhost" || host == "127.0.0.1" || host == "::1" { + return true + } + ip := net.ParseIP(host) + if ip != nil && ip.IsLoopback() { + return true + } + ifaces, err := net.InterfaceAddrs() + if err != nil { + return false + } + for _, ia := range ifaces { + var got net.IP + switch v := ia.(type) { + case *net.IPNet: + got = v.IP + case *net.IPAddr: + got = v.IP + } + if got == nil { + continue + } + if strings.EqualFold(got.String(), host) { + return true + } + if ip != nil && got.Equal(ip) { + return true + } + } + return false +} diff --git a/weed/util/net_host_test.go b/weed/util/net_host_test.go new file mode 100644 index 0000000000..7caf30f056 --- /dev/null +++ b/weed/util/net_host_test.go @@ -0,0 +1,24 @@ +package util + +import "testing" + +func TestHostKeyStripsPort(t *testing.T) { + t.Parallel() + if got := HostKey("10.87.0.3:8080"); got != "10.87.0.3" { + t.Fatalf("HostKey replica: got %q", got) + } + if got := HostKey("10.87.0.3:33646"); got != "10.87.0.3" { + t.Fatalf("HostKey worker: got %q", got) + } + if HostKey("10.87.0.3:8080") != HostKey("10.87.0.3:33646") { + t.Fatal("worker and volume server on the same IP must share a HostKey") + } +} + +func TestHostKeysFromAddressesDedups(t *testing.T) { + t.Parallel() + got := HostKeysFromAddresses([]string{"10.0.0.1:8080", "10.0.0.1:8081", "10.0.0.2:8080", ""}) + if len(got) != 2 || got[0] != "10.0.0.1" || got[1] != "10.0.0.2" { + t.Fatalf("got %v", got) + } +} diff --git a/weed/worker/tasks/erasure_coding/ec_task.go b/weed/worker/tasks/erasure_coding/ec_task.go index b4ba0f126b..ab56c8e997 100644 --- a/weed/worker/tasks/erasure_coding/ec_task.go +++ b/weed/worker/tasks/erasure_coding/ec_task.go @@ -21,6 +21,7 @@ import ( storagetypes "github.com/seaweedfs/seaweedfs/weed/storage/types" "github.com/seaweedfs/seaweedfs/weed/storage/volume_info" "github.com/seaweedfs/seaweedfs/weed/storage/volume_replica" + "github.com/seaweedfs/seaweedfs/weed/util" "github.com/seaweedfs/seaweedfs/weed/wdclient" "github.com/seaweedfs/seaweedfs/weed/worker/types" "github.com/seaweedfs/seaweedfs/weed/worker/types/base" @@ -403,9 +404,52 @@ func (t *ErasureCodingTask) syncAndSelectSourceReplica() error { glog.Infof("EC encode: using best replica %s as source for volume %d (was %s)", best.Url, t.volumeID, t.server) t.server = best.Url } + // Host-affinity: if this worker sits on a replica and that copy is as + // complete as the union (typical 010 after sync), pull from localhost + // instead of the lexically-first / largest remote replica. + if local := preferLocalReplicaIfComplete(t.grpcDialOption, needle.VolumeId(t.volumeID), locs, t.server); local != "" && local != t.server { + glog.Infof("EC encode: preferring local replica %s as source for volume %d (was %s)", local, t.volumeID, t.server) + t.server = local + } return nil } +// preferLocalReplicaIfComplete returns a replica URL on this machine whose +// file count is >= the best remote replica. After SyncAndSelectBestReplica +// a consistent set is equal everywhere; an inconsistent union lives only on +// the chosen best, so a stale local copy is not selected. +func preferLocalReplicaIfComplete(dial grpc.DialOption, vid needle.VolumeId, locs []wdclient.Location, current string) string { + var localURL string + for _, loc := range locs { + if util.AddressIsLocal(loc.Url) { + localURL = loc.Url + break + } + } + if localURL == "" || localURL == current { + return current + } + statuses := volume_replica.GetReplicaStatuses(dial, vid, locs) + var localCount, maxCount uint64 + var sawLocal bool + for _, s := range statuses { + if s.Error != nil { + continue + } + if s.FileCount > maxCount { + maxCount = s.FileCount + } + if s.Location.Url == localURL { + localCount = s.FileCount + sawLocal = true + } + } + if !sawLocal || localCount < maxCount { + return current + } + return localURL +} + // rollbackReadonly is a best-effort restore of every replica markReplicasReadonly // touched, used when the EC task fails before the originals are deleted. Logs but // does not return errors; uses a fresh context since the caller's may be cancelled. diff --git a/weed/worker/tasks/erasure_coding/plugin_handler.go b/weed/worker/tasks/erasure_coding/plugin_handler.go index 1b8f74585e..f43250ebd2 100644 --- a/weed/worker/tasks/erasure_coding/plugin_handler.go +++ b/weed/worker/tasks/erasure_coding/plugin_handler.go @@ -670,8 +670,15 @@ func buildErasureCodingProposal( } sourceNode := "" - if len(params.Sources) > 0 { - sourceNode = strings.TrimSpace(params.Sources[0].Node) + var replicaServers []string + for _, s := range params.Sources { + if s == nil || strings.TrimSpace(s.Node) == "" || len(s.ShardIds) > 0 { + continue + } + replicaServers = append(replicaServers, s.Node) + if sourceNode == "" { + sourceNode = strings.TrimSpace(s.Node) + } } summary := fmt.Sprintf("Erasure code volume %d", result.VolumeID) @@ -702,6 +709,9 @@ func buildErasureCodingProposal( "source_server": { Kind: &plugin_pb.ConfigValue_StringValue{StringValue: sourceNode}, }, + "replica_servers": { + Kind: &plugin_pb.ConfigValue_StringList{StringList: &plugin_pb.StringList{Values: replicaServers}}, + }, "collection": { Kind: &plugin_pb.ConfigValue_StringValue{StringValue: result.Collection}, },