mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-08-16 04:06:44 +00:00
Chris LuandGitHub
d1665750e1
Delete the EC placement package and the dead encode planner code Now that encode (and repair) place via ecbalancer.Place, nothing uses the erasure_coding/placement package or the EC-only planner machinery (ecPlacementPlanner, diskInfosToCandidates, calculateECScoreCandidate, distributeECShards) in detection.go. Removes them and the package, along with the planner-direct unit tests.
738 lines
30 KiB
Go
738 lines
30 KiB
Go
package erasure_coding
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"sort"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/seaweedfs/seaweedfs/weed/admin/topology"
|
|
"github.com/seaweedfs/seaweedfs/weed/glog"
|
|
"github.com/seaweedfs/seaweedfs/weed/operation"
|
|
"github.com/seaweedfs/seaweedfs/weed/pb"
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/volume_server_pb"
|
|
"github.com/seaweedfs/seaweedfs/weed/pb/worker_pb"
|
|
"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding"
|
|
"github.com/seaweedfs/seaweedfs/weed/storage/erasure_coding/ecbalancer"
|
|
"github.com/seaweedfs/seaweedfs/weed/storage/super_block"
|
|
"github.com/seaweedfs/seaweedfs/weed/util/wildcard"
|
|
"github.com/seaweedfs/seaweedfs/weed/worker/tasks/base"
|
|
workerutil "github.com/seaweedfs/seaweedfs/weed/worker/tasks/util"
|
|
"github.com/seaweedfs/seaweedfs/weed/worker/types"
|
|
)
|
|
|
|
const (
|
|
minProposalsBeforeEarlyStop = 10
|
|
maxConsecutivePlanningFailures = 10
|
|
)
|
|
|
|
// Detection implements the detection logic for erasure coding tasks.
|
|
// It respects ctx cancellation and can stop early once maxResults is reached.
|
|
func Detection(ctx context.Context, metrics []*types.VolumeHealthMetrics, clusterInfo *types.ClusterInfo, config base.TaskConfig, maxResults int) ([]*types.TaskDetectionResult, bool, error) {
|
|
if !config.IsEnabled() {
|
|
return nil, false, nil
|
|
}
|
|
|
|
if maxResults < 0 {
|
|
maxResults = 0
|
|
}
|
|
|
|
ecConfig := config.(*Config)
|
|
var results []*types.TaskDetectionResult
|
|
hasMore := false
|
|
stoppedEarly := false
|
|
now := time.Now()
|
|
quietThreshold := time.Duration(ecConfig.QuietForSeconds) * time.Second
|
|
minSizeBytes := uint64(ecConfig.MinSizeMB) * 1024 * 1024 // Configurable minimum
|
|
|
|
debugCount := 0
|
|
skippedAlreadyEC := 0
|
|
skippedTooSmall := 0
|
|
skippedCollectionFilter := 0
|
|
skippedQuietTime := 0
|
|
skippedFullness := 0
|
|
skippedRemote := 0
|
|
skippedTooFewNodes := 0
|
|
consecutivePlanningFailures := 0
|
|
|
|
// EC shard replica placement: explicit config wins, else the master default.
|
|
var replicaPlacement *super_block.ReplicaPlacement
|
|
if clusterInfo != nil {
|
|
replicaPlacement = super_block.ResolveReplicaPlacement(ecConfig.ReplicaPlacement, clusterInfo.DefaultReplicaPlacement)
|
|
}
|
|
// EC placement honors only the rack/node digits; the data-center digit can't
|
|
// express a useful per-DC EC shard cap (it maxes at 2). Warn once per cycle so a
|
|
// 1xx/2xx setting isn't silently ineffective.
|
|
if replicaPlacement != nil && replicaPlacement.DiffDataCenterCount > 0 {
|
|
glog.Warningf("EC Detection: replica placement data-center digit (%d) is ignored for EC; only rack/node digits are honored", replicaPlacement.DiffDataCenterCount)
|
|
}
|
|
|
|
allowedCollections := wildcard.CompileWildcardMatchers(ecConfig.CollectionFilter)
|
|
|
|
// Cluster node count for the min-node safety gate (mirrors the shell ec.encode
|
|
// guard that refuses to encode when nodes < parity shards, so shards cannot be
|
|
// spread for fault tolerance).
|
|
clusterNodeCount := countTopologyNodes(clusterInfo.ActiveTopology)
|
|
|
|
// Group metrics by VolumeID to handle replicas and select canonical server
|
|
volumeGroups := make(map[uint32][]*types.VolumeHealthMetrics)
|
|
for _, metric := range metrics {
|
|
if ctx != nil {
|
|
if err := ctx.Err(); err != nil {
|
|
return results, hasMore, err
|
|
}
|
|
}
|
|
volumeGroups[metric.VolumeID] = append(volumeGroups[metric.VolumeID], metric)
|
|
}
|
|
|
|
groupKeys := make([]uint32, 0, len(volumeGroups))
|
|
for volumeID := range volumeGroups {
|
|
groupKeys = append(groupKeys, volumeID)
|
|
}
|
|
sort.Slice(groupKeys, func(i, j int) bool { return groupKeys[i] < groupKeys[j] })
|
|
|
|
// Iterate over groups to check criteria and creation tasks
|
|
for idx, volumeID := range groupKeys {
|
|
if ctx != nil {
|
|
if err := ctx.Err(); err != nil {
|
|
return results, hasMore, err
|
|
}
|
|
}
|
|
if maxResults > 0 && len(results) >= maxResults {
|
|
if idx+1 < len(groupKeys) {
|
|
hasMore = true
|
|
}
|
|
stoppedEarly = true
|
|
break
|
|
}
|
|
|
|
groupMetrics := volumeGroups[volumeID]
|
|
|
|
// Find canonical metric (lowest Server ID) to ensure consistent task deduplication
|
|
metric := groupMetrics[0]
|
|
for _, m := range groupMetrics {
|
|
if m.Server < metric.Server {
|
|
metric = m
|
|
}
|
|
}
|
|
|
|
// Skip if already EC volume
|
|
if metric.IsECVolume {
|
|
skippedAlreadyEC++
|
|
continue
|
|
}
|
|
|
|
// Handle the "stuck source" state from #9448: a previous encode
|
|
// succeeded but the post-encode source-delete left a regular replica
|
|
// behind, so the master heartbeats BOTH the replica AND its EC shards.
|
|
// metric.IsECVolume above is set only for the EC-side metric path, so
|
|
// the canonical metric we picked is the regular replica with
|
|
// IsECVolume=false. Re-proposing an encode in that state collides with
|
|
// the mounted shards on the targets ("ec volume %d is mounted; refusing
|
|
// overwrite") and the detector re-queues forever.
|
|
//
|
|
// We only act when the EC shard set is COMPLETE — fewer than
|
|
// totalShards present means the existing recovery branch below
|
|
// (around the `existingECShards` block) should keep its chance to
|
|
// fold the partial shards into the new task. Counting walks
|
|
// EcIndexBits to handle a single info entry carrying multiple shards.
|
|
if clusterInfo.ActiveTopology != nil {
|
|
shardCount := countExistingEcShardsForVolume(clusterInfo.ActiveTopology, metric.VolumeID, metric.Collection)
|
|
totalShards := erasure_coding.DataShardsCount + erasure_coding.ParityShardsCount
|
|
if shardCount >= totalShards {
|
|
glog.Warningf("EC Detection: Volume %d has all %d EC shards in topology; "+
|
|
"source replica on %s is orphaned (#9448).",
|
|
metric.VolumeID, totalShards, metric.Server)
|
|
if clusterInfo.GrpcDialOption != nil {
|
|
deleted, cleanupErr := cleanupOrphanSourceReplicas(ctx, clusterInfo, metric, totalShards)
|
|
switch {
|
|
case cleanupErr != nil:
|
|
// Don't fall through to a re-encode — that would just
|
|
// collide with the mounted shards again. Surface the
|
|
// failure and wait for the next cycle; the source is
|
|
// still safe.
|
|
glog.Warningf("EC Detection: failed to auto-clean orphaned source for volume %d: %v", metric.VolumeID, cleanupErr)
|
|
case deleted > 0:
|
|
glog.Infof("EC Detection: auto-cleaned %d orphaned source replica(s) for volume %d after verifying all %d EC shards present", deleted, metric.VolumeID, totalShards)
|
|
default:
|
|
glog.V(1).Infof("EC Detection: no orphaned regular replicas found in topology for volume %d (collection %q)", metric.VolumeID, metric.Collection)
|
|
}
|
|
} else {
|
|
glog.Warningf("EC Detection: no gRPC dial option available to auto-clean orphaned source for volume %d; "+
|
|
"to clean up by hand, send a targeted VolumeDelete RPC to %s only — DO NOT use the cluster-wide `volume.delete` shell command, which would also delete the EC shards.",
|
|
metric.VolumeID, metric.Server)
|
|
}
|
|
skippedAlreadyEC++
|
|
continue
|
|
}
|
|
}
|
|
|
|
// Check minimum size requirement
|
|
if metric.Size < minSizeBytes {
|
|
skippedTooSmall++
|
|
continue
|
|
}
|
|
|
|
// Check collection filter if specified
|
|
if len(allowedCollections) > 0 && !wildcard.MatchesAnyWildcard(allowedCollections, metric.Collection) {
|
|
skippedCollectionFilter++
|
|
continue
|
|
}
|
|
|
|
// Skip remote/tiered volumes: encoding them would lose the tiering. The
|
|
// shell ec.encode excludes remote volumes for the same reason.
|
|
if metric.HasRemoteCopy {
|
|
skippedRemote++
|
|
continue
|
|
}
|
|
|
|
// Min-node safety gate: don't encode when the cluster has fewer nodes than
|
|
// this collection's parity shards — shards could not be spread to tolerate
|
|
// failures. Mirrors the shell ec.encode guard (node count < parity shards).
|
|
if clusterNodeCount > 0 && clusterNodeCount < erasure_coding.ParityShardsCount {
|
|
skippedTooFewNodes++
|
|
continue
|
|
}
|
|
|
|
// Check quiet duration and fullness criteria
|
|
if metric.Age >= quietThreshold && metric.FullnessRatio >= ecConfig.FullnessRatio {
|
|
if ctx != nil {
|
|
if err := ctx.Err(); err != nil {
|
|
return results, hasMore, err
|
|
}
|
|
}
|
|
glog.Infof("EC Detection: Volume %d meets all criteria, attempting to create task", metric.VolumeID)
|
|
|
|
// Generate task ID for ActiveTopology integration
|
|
taskID := fmt.Sprintf("ec_vol_%d_%d", metric.VolumeID, now.Unix())
|
|
|
|
result := &types.TaskDetectionResult{
|
|
TaskID: taskID, // Link to ActiveTopology pending task
|
|
TaskType: types.TaskTypeErasureCoding,
|
|
VolumeID: metric.VolumeID,
|
|
Server: metric.Server,
|
|
Collection: metric.Collection,
|
|
Priority: types.TaskPriorityLow, // EC is not urgent
|
|
Reason: fmt.Sprintf("Volume meets EC criteria: quiet for %.1fs (>%ds), fullness=%.1f%% (>%.1f%%), size=%.1fMB (>%dMB)",
|
|
metric.Age.Seconds(), ecConfig.QuietForSeconds, metric.FullnessRatio*100, ecConfig.FullnessRatio*100,
|
|
float64(metric.Size)/(1024*1024), ecConfig.MinSizeMB),
|
|
ScheduleAt: now,
|
|
}
|
|
|
|
// Plan EC destinations if ActiveTopology is available
|
|
if clusterInfo.ActiveTopology != nil {
|
|
// Check if ANY task already exists in ActiveTopology for this volume
|
|
if clusterInfo.ActiveTopology.HasAnyTask(metric.VolumeID) {
|
|
glog.V(2).Infof("EC Detection: Skipping volume %d, task already exists in ActiveTopology", metric.VolumeID)
|
|
continue
|
|
}
|
|
|
|
glog.Infof("EC Detection: ActiveTopology available, planning destinations for volume %d", metric.VolumeID)
|
|
dataShards := erasure_coding.DataShardsCount
|
|
parityShards := erasure_coding.ParityShardsCount
|
|
multiPlan, shardsPerPlan, err := planECDestinations(clusterInfo.ActiveTopology, metric, ecConfig, replicaPlacement, dataShards, parityShards)
|
|
if err != nil {
|
|
glog.V(2).Infof("Failed to plan EC destinations for volume %d: %v", metric.VolumeID, err)
|
|
consecutivePlanningFailures++
|
|
if len(results) >= minProposalsBeforeEarlyStop && consecutivePlanningFailures >= maxConsecutivePlanningFailures {
|
|
glog.Warningf("EC Detection: stopping early after %d consecutive placement failures with %d proposals already planned", consecutivePlanningFailures, len(results))
|
|
hasMore = true
|
|
stoppedEarly = true
|
|
break
|
|
}
|
|
continue // Skip this volume if destination planning fails
|
|
}
|
|
consecutivePlanningFailures = 0
|
|
glog.Infof("EC Detection: Successfully planned %d destinations for volume %d", len(multiPlan.Plans), metric.VolumeID)
|
|
|
|
// Calculate expected shard size for EC operation
|
|
// Each data shard will be approximately volumeSize / dataShards
|
|
expectedShardSize := uint64(metric.Size) / uint64(dataShards)
|
|
|
|
// Add pending EC shard task to ActiveTopology for capacity management
|
|
|
|
// Extract shard destinations from multiPlan
|
|
var shardDestinations []string
|
|
var shardDiskIDs []uint32
|
|
for _, plan := range multiPlan.Plans {
|
|
shardDestinations = append(shardDestinations, plan.TargetNode)
|
|
shardDiskIDs = append(shardDiskIDs, plan.TargetDisk)
|
|
}
|
|
|
|
// Find all volume replica locations (server + disk) from topology
|
|
glog.Infof("EC Detection: Looking for replica locations for volume %d", metric.VolumeID)
|
|
replicaLocations := findVolumeReplicaLocations(clusterInfo.ActiveTopology, metric.VolumeID, metric.Collection)
|
|
if len(replicaLocations) == 0 {
|
|
glog.Warningf("No replica locations found for volume %d, skipping EC", metric.VolumeID)
|
|
continue
|
|
}
|
|
glog.Infof("EC Detection: Found %d replica locations for volume %d", len(replicaLocations), metric.VolumeID)
|
|
|
|
// Find existing EC shards from previous failed attempts
|
|
existingECShards := findExistingECShards(clusterInfo.ActiveTopology, metric.VolumeID, metric.Collection)
|
|
|
|
// Combine volume replicas and existing EC shards for cleanup
|
|
var sources []topology.TaskSourceSpec
|
|
|
|
// Add volume replicas (will free volume slots)
|
|
for _, replica := range replicaLocations {
|
|
sources = append(sources, topology.TaskSourceSpec{
|
|
ServerID: replica.ServerID,
|
|
DiskID: replica.DiskID,
|
|
DataCenter: replica.DataCenter,
|
|
Rack: replica.Rack,
|
|
CleanupType: topology.CleanupVolumeReplica,
|
|
})
|
|
}
|
|
|
|
// Add existing EC shards (will free shard slots)
|
|
duplicateCheck := make(map[string]bool)
|
|
for _, replica := range replicaLocations {
|
|
key := fmt.Sprintf("%s:%d", replica.ServerID, replica.DiskID)
|
|
duplicateCheck[key] = true
|
|
}
|
|
|
|
for _, shard := range existingECShards {
|
|
key := fmt.Sprintf("%s:%d", shard.ServerID, shard.DiskID)
|
|
if !duplicateCheck[key] { // Avoid duplicates if EC shards are on same disk as volume replicas
|
|
sources = append(sources, topology.TaskSourceSpec{
|
|
ServerID: shard.ServerID,
|
|
DiskID: shard.DiskID,
|
|
DataCenter: shard.DataCenter,
|
|
Rack: shard.Rack,
|
|
CleanupType: topology.CleanupECShards,
|
|
ShardIds: append([]uint32(nil), shard.ShardIds...),
|
|
})
|
|
duplicateCheck[key] = true
|
|
}
|
|
}
|
|
|
|
glog.V(2).Infof("Found %d volume replicas and %d existing EC shards for volume %d (total %d cleanup sources)",
|
|
len(replicaLocations), len(existingECShards), metric.VolumeID, len(sources))
|
|
|
|
// Convert shard destinations to TaskDestinationSpec. A destination may
|
|
// hold several shards (small clusters), so reserve capacity for the
|
|
// actual per-disk shard count that Place assigned (shardsPerPlan),
|
|
// which is exactly what createECTargets writes.
|
|
destinations := make([]topology.TaskDestinationSpec, len(shardDestinations))
|
|
for i, dest := range shardDestinations {
|
|
shardCount := len(shardsPerPlan[i])
|
|
shardImpact := topology.CalculateECShardStorageImpact(int32(shardCount), int64(expectedShardSize))
|
|
destSize := int64(expectedShardSize) * int64(shardCount)
|
|
destinations[i] = topology.TaskDestinationSpec{
|
|
ServerID: dest,
|
|
DiskID: shardDiskIDs[i],
|
|
StorageImpact: &shardImpact,
|
|
EstimatedSize: &destSize,
|
|
}
|
|
}
|
|
|
|
// Convert sources before mutating topology
|
|
sourcesProto, err := convertTaskSourcesToProtobuf(sources, metric.VolumeID, clusterInfo.ActiveTopology)
|
|
if err != nil {
|
|
glog.Warningf("Failed to convert sources for EC task on volume %d: %v, skipping", metric.VolumeID, err)
|
|
continue
|
|
}
|
|
|
|
err = clusterInfo.ActiveTopology.AddPendingTask(topology.TaskSpec{
|
|
TaskID: taskID,
|
|
TaskType: topology.TaskTypeErasureCoding,
|
|
VolumeID: metric.VolumeID,
|
|
VolumeSize: int64(metric.Size),
|
|
Sources: sources,
|
|
Destinations: destinations,
|
|
})
|
|
if err != nil {
|
|
glog.Warningf("Failed to add pending EC shard task to ActiveTopology for volume %d: %v", metric.VolumeID, err)
|
|
continue // Skip this volume if topology task addition fails
|
|
}
|
|
|
|
// Cross-volume in-cycle capacity is tracked by ActiveTopology via the
|
|
// pending task above, which the next volume's FromActiveTopology snapshot
|
|
// reflects; no separate planner reservation is needed.
|
|
|
|
glog.V(2).Infof("Added pending EC shard task %s to ActiveTopology for volume %d with %d cleanup sources and %d shard destinations",
|
|
taskID, metric.VolumeID, len(sources), len(multiPlan.Plans))
|
|
|
|
// Create unified sources and targets for EC task
|
|
result.TypedParams = &worker_pb.TaskParams{
|
|
TaskId: taskID, // Link to ActiveTopology pending task
|
|
VolumeId: metric.VolumeID,
|
|
Collection: metric.Collection,
|
|
VolumeSize: metric.Size, // Store original volume size for tracking changes
|
|
|
|
// Unified sources - all sources that will be processed/cleaned up
|
|
Sources: sourcesProto,
|
|
|
|
// Unified targets - all EC shard destinations
|
|
Targets: createECTargets(multiPlan, shardsPerPlan),
|
|
|
|
TaskParams: &worker_pb.TaskParams_ErasureCodingParams{
|
|
ErasureCodingParams: createECTaskParams(dataShards, parityShards, metric.DiskType),
|
|
},
|
|
}
|
|
|
|
glog.V(1).Infof("Planned EC destinations for volume %d: %d shards across %d racks, %d DCs",
|
|
metric.VolumeID, len(multiPlan.Plans), multiPlan.SuccessfulRack, multiPlan.SuccessfulDCs)
|
|
} else {
|
|
glog.Warningf("No ActiveTopology available for destination planning in EC detection")
|
|
continue // Skip this volume if no topology available
|
|
}
|
|
|
|
glog.Infof("EC Detection: Successfully created EC task for volume %d, adding to results", metric.VolumeID)
|
|
results = append(results, result)
|
|
} else {
|
|
// Count debug reasons
|
|
if metric.Age < quietThreshold {
|
|
skippedQuietTime++
|
|
}
|
|
if metric.FullnessRatio < ecConfig.FullnessRatio {
|
|
skippedFullness++
|
|
}
|
|
|
|
if debugCount < 5 { // Limit to avoid spam
|
|
// Logic moved outside
|
|
}
|
|
debugCount++
|
|
}
|
|
}
|
|
|
|
// Log debug summary if no tasks were created
|
|
if len(results) == 0 && len(metrics) > 0 && !stoppedEarly {
|
|
totalVolumes := len(metrics)
|
|
glog.V(1).Infof("EC detection: No tasks created for %d volumes (skipped: %d already EC, %d too small, %d filtered, %d not quiet, %d not full, %d remote, %d too few nodes)",
|
|
totalVolumes, skippedAlreadyEC, skippedTooSmall, skippedCollectionFilter, skippedQuietTime, skippedFullness, skippedRemote, skippedTooFewNodes)
|
|
|
|
// Show details for first few volumes
|
|
for i, metric := range metrics {
|
|
if i >= 3 || metric.IsECVolume { // Limit to first 3 non-EC volumes
|
|
continue
|
|
}
|
|
sizeMB := float64(metric.Size) / (1024 * 1024)
|
|
glog.V(1).Infof("ERASURE CODING: Volume %d: size=%.1fMB (need ≥%dMB), age=%s (need ≥%s), fullness=%.1f%% (need ≥%.1f%%)",
|
|
metric.VolumeID, sizeMB, ecConfig.MinSizeMB, metric.Age.Truncate(time.Minute), quietThreshold.Truncate(time.Minute),
|
|
metric.FullnessRatio*100, ecConfig.FullnessRatio*100)
|
|
}
|
|
}
|
|
|
|
return results, hasMore, nil
|
|
}
|
|
|
|
// countTopologyNodes counts volume-server nodes in the active topology, used by
|
|
// the min-node safety gate.
|
|
func countTopologyNodes(at *topology.ActiveTopology) int {
|
|
if at == nil {
|
|
return 0
|
|
}
|
|
topo := at.GetTopologyInfo()
|
|
if topo == nil {
|
|
return 0
|
|
}
|
|
n := 0
|
|
for _, dc := range topo.DataCenterInfos {
|
|
for _, rack := range dc.RackInfos {
|
|
n += len(rack.DataNodeInfos)
|
|
}
|
|
}
|
|
return n
|
|
}
|
|
|
|
// planECDestinations places all shards of the volume via the shared ecbalancer
|
|
// policy and returns the per-disk destination plans plus, parallel to them, the
|
|
// shard ids ecbalancer.Place assigned to each disk (so createECTargets and the
|
|
// capacity reservations use the real assignment, not a round-robin guess).
|
|
//
|
|
// Encode is lenient (PlaceDurabilityFirst): it relaxes caps/anti-affinity/RP as
|
|
// needed rather than fail, and prefers the source disk type but spills if that
|
|
// type can't hold every shard. rp is the resolved replica placement (may be nil).
|
|
func planECDestinations(at *topology.ActiveTopology, metric *types.VolumeHealthMetrics, ecConfig *Config, rp *super_block.ReplicaPlacement, dataShards, parityShards int) (*topology.MultiDestinationPlan, [][]uint32, error) {
|
|
if at == nil {
|
|
return nil, nil, fmt.Errorf("active topology not available for EC placement")
|
|
}
|
|
if dataShards <= 0 || parityShards <= 0 {
|
|
return nil, nil, fmt.Errorf("invalid EC ratio: dataShards=%d parityShards=%d", dataShards, parityShards)
|
|
}
|
|
totalShards := dataShards + parityShards
|
|
// Survive losing one disk: each disk holds at most parityShards shards,
|
|
// so we need at least ceil(totalShards / parityShards) disks.
|
|
minTotalDisks := (totalShards + parityShards - 1) / parityShards
|
|
expectedShardSize := uint64(metric.Size) / uint64(dataShards)
|
|
|
|
snap := ecbalancer.FromActiveTopology(at, dataShards)
|
|
// Encode is greenfield: any EC shards already present for this volume are stale
|
|
// leftovers from a prior failed attempt, which the task deletes
|
|
// (cleanupStaleEcShards) before distributing the new shards. Release them so they
|
|
// don't occupy capacity or skew anti-affinity / per-disk caps during planning.
|
|
snap.ReleaseVolumeShards(metric.Collection, metric.VolumeID)
|
|
need := make([]int, totalShards)
|
|
for i := range need {
|
|
need[i] = i
|
|
}
|
|
res, err := snap.Place(metric.VolumeID, metric.Collection, need, ecbalancer.Constraints{
|
|
DiskType: metric.DiskType,
|
|
DiskTypePolicy: ecbalancer.DiskTypePrefer,
|
|
PreferredTags: ecConfig.PreferredTags,
|
|
ReplicaPlacement: rp,
|
|
Ratio: func(string) (int, int) { return dataShards, parityShards },
|
|
}, ecbalancer.PlaceDurabilityFirst)
|
|
if err != nil {
|
|
return nil, nil, err
|
|
}
|
|
if res.SpilledToOtherDiskType {
|
|
glog.Warningf("EC volume %d: placed shards outside preferred disk type %q", metric.VolumeID, metric.DiskType)
|
|
}
|
|
if res.SpilledOutsidePreferredTags {
|
|
glog.Warningf("EC volume %d: placed shards outside preferred tags %v", metric.VolumeID, ecConfig.PreferredTags)
|
|
}
|
|
if len(res.Relaxed) > 0 {
|
|
// Encode is best-effort (PlaceDurabilityFirst): it relaxes these constraints
|
|
// rather than defer when the cluster can't satisfy them. Surface it so a tight
|
|
// replica placement isn't silently weakened; rebalancing tightens the spread.
|
|
glog.Warningf("EC volume %d: placed with relaxed constraints %v; replica placement not fully satisfied (rebalancing will adjust)", metric.VolumeID, res.Relaxed)
|
|
}
|
|
|
|
// Group the per-shard destinations into one plan per (node,disk), iterating
|
|
// shard ids in order for determinism.
|
|
type diskGroup struct {
|
|
node, rack, dc string
|
|
diskID uint32
|
|
shards []uint32
|
|
}
|
|
type diskKey struct {
|
|
node string
|
|
diskID uint32
|
|
}
|
|
groups := make(map[diskKey]*diskGroup, totalShards)
|
|
order := make([]diskKey, 0, totalShards)
|
|
for sid := 0; sid < totalShards; sid++ {
|
|
d, ok := res.Destinations[sid]
|
|
if !ok {
|
|
return nil, nil, fmt.Errorf("EC volume %d: shard %d was not placed", metric.VolumeID, sid)
|
|
}
|
|
key := diskKey{node: d.Node, diskID: d.DiskID}
|
|
g := groups[key]
|
|
if g == nil {
|
|
g = &diskGroup{node: d.Node, rack: d.Rack, dc: d.DataCenter, diskID: d.DiskID}
|
|
groups[key] = g
|
|
order = append(order, key)
|
|
}
|
|
g.shards = append(g.shards, uint32(sid))
|
|
}
|
|
if len(order) < minTotalDisks {
|
|
return nil, nil, fmt.Errorf("placed onto %d disks, but EC %d+%d needs at least %d so no disk holds more than %d shards",
|
|
len(order), dataShards, parityShards, minTotalDisks, parityShards)
|
|
}
|
|
|
|
var plans []*topology.DestinationPlan
|
|
shardsPerPlan := make([][]uint32, 0, len(order))
|
|
rackCount := make(map[string]int)
|
|
dcCount := make(map[string]int)
|
|
for _, key := range order {
|
|
g := groups[key]
|
|
targetAddress, err := workerutil.ResolveServerAddress(g.node, at)
|
|
if err != nil {
|
|
return nil, nil, fmt.Errorf("failed to resolve address for target server %s: %v", g.node, err)
|
|
}
|
|
plans = append(plans, &topology.DestinationPlan{
|
|
TargetNode: g.node,
|
|
TargetAddress: targetAddress,
|
|
TargetDisk: g.diskID,
|
|
TargetRack: g.rack,
|
|
TargetDC: g.dc,
|
|
ExpectedSize: expectedShardSize,
|
|
})
|
|
shardsPerPlan = append(shardsPerPlan, g.shards)
|
|
rackCount[fmt.Sprintf("%s:%s", g.dc, g.rack)]++
|
|
dcCount[g.dc]++
|
|
}
|
|
|
|
glog.V(1).Infof("Planned EC destinations for volume %d (size=%d bytes): expected shard size=%d bytes, %d shards across %d disks, %d racks, %d DCs",
|
|
metric.VolumeID, metric.Size, expectedShardSize, totalShards, len(plans), len(rackCount), len(dcCount))
|
|
|
|
return &topology.MultiDestinationPlan{
|
|
Plans: plans,
|
|
TotalShards: totalShards,
|
|
SuccessfulRack: len(rackCount),
|
|
SuccessfulDCs: len(dcCount),
|
|
}, shardsPerPlan, nil
|
|
}
|
|
|
|
// createECTargets builds TaskTargets from the per-disk plans and the shard ids
|
|
// ecbalancer.Place assigned to each (shardsPerPlan is parallel to multiPlan.Plans).
|
|
func createECTargets(multiPlan *topology.MultiDestinationPlan, shardsPerPlan [][]uint32) []*worker_pb.TaskTarget {
|
|
targets := make([]*worker_pb.TaskTarget, 0, len(multiPlan.Plans))
|
|
for i, plan := range multiPlan.Plans {
|
|
shardIDs := shardsPerPlan[i]
|
|
targets = append(targets, &worker_pb.TaskTarget{
|
|
Node: plan.TargetAddress,
|
|
DiskId: plan.TargetDisk,
|
|
Rack: plan.TargetRack,
|
|
DataCenter: plan.TargetDC,
|
|
ShardIds: shardIDs,
|
|
EstimatedSize: plan.ExpectedSize,
|
|
})
|
|
glog.V(2).Infof("EC planning: target %s disk %d assigned shards %v", plan.TargetNode, plan.TargetDisk, shardIDs)
|
|
}
|
|
return targets
|
|
}
|
|
|
|
// convertTaskSourcesToProtobuf converts topology.TaskSourceSpec to worker_pb.TaskSource
|
|
func convertTaskSourcesToProtobuf(sources []topology.TaskSourceSpec, volumeID uint32, activeTopology *topology.ActiveTopology) ([]*worker_pb.TaskSource, error) {
|
|
var protobufSources []*worker_pb.TaskSource
|
|
|
|
for _, source := range sources {
|
|
serverAddress, err := workerutil.ResolveServerAddress(source.ServerID, activeTopology)
|
|
if err != nil {
|
|
return nil, fmt.Errorf("failed to resolve address for source server %s: %v", source.ServerID, err)
|
|
}
|
|
|
|
pbSource := &worker_pb.TaskSource{
|
|
Node: serverAddress,
|
|
DiskId: source.DiskID,
|
|
DataCenter: source.DataCenter,
|
|
Rack: source.Rack,
|
|
}
|
|
|
|
// Convert storage impact to estimated size
|
|
if source.EstimatedSize != nil {
|
|
pbSource.EstimatedSize = uint64(*source.EstimatedSize)
|
|
}
|
|
|
|
// Populated ShardIds is the wire-level marker that flags an
|
|
// EC-shard cleanup source; the worker routes it through
|
|
// cleanupStaleEcShards and skips it in getReplicas.
|
|
switch source.CleanupType {
|
|
case topology.CleanupVolumeReplica:
|
|
pbSource.VolumeId = volumeID
|
|
case topology.CleanupECShards:
|
|
pbSource.VolumeId = volumeID
|
|
pbSource.ShardIds = append([]uint32(nil), source.ShardIds...)
|
|
}
|
|
|
|
protobufSources = append(protobufSources, pbSource)
|
|
}
|
|
|
|
return protobufSources, nil
|
|
}
|
|
|
|
// createECTaskParams creates clean EC task parameters (destinations now in unified targets).
|
|
// sourceDiskType is forwarded to VolumeEcShardsMount so the resulting EC volume
|
|
// reports under the source's disk type rather than the target location's (#9423).
|
|
func createECTaskParams(dataShards, parityShards int, sourceDiskType string) *worker_pb.ErasureCodingTaskParams {
|
|
return &worker_pb.ErasureCodingTaskParams{
|
|
DataShards: int32(dataShards),
|
|
ParityShards: int32(parityShards),
|
|
SourceDiskType: sourceDiskType,
|
|
}
|
|
}
|
|
|
|
// findVolumeReplicaLocations finds all replica locations (server + disk) for the specified volume
|
|
// Uses O(1) indexed lookup for optimal performance on large clusters.
|
|
func findVolumeReplicaLocations(activeTopology *topology.ActiveTopology, volumeID uint32, collection string) []topology.VolumeReplica {
|
|
if activeTopology == nil {
|
|
return nil
|
|
}
|
|
return activeTopology.GetVolumeLocations(volumeID, collection)
|
|
}
|
|
|
|
// findExistingECShards finds existing EC shards for a volume (from previous failed EC attempts)
|
|
// Uses O(1) indexed lookup for optimal performance on large clusters.
|
|
func findExistingECShards(activeTopology *topology.ActiveTopology, volumeID uint32, collection string) []topology.VolumeReplica {
|
|
if activeTopology == nil {
|
|
return nil
|
|
}
|
|
return activeTopology.GetECShardLocations(volumeID, collection)
|
|
}
|
|
|
|
// cleanupOrphanSourceReplicas deletes any regular volume replicas still
|
|
// present in the topology for (volumeID, collection) after re-verifying that
|
|
// the full EC shard set is intact. Caller must hold expectedShards equal to
|
|
// the configured totalShards count. Issues VolumeDelete RPC to each replica
|
|
// server's address — that endpoint only touches the regular volume on the
|
|
// targeted server, never EC shards (those live in a separate store path).
|
|
// The cluster-wide `volume.delete` shell command is what would have nuked
|
|
// the EC shards too; the targeted RPC used here is safe by construction.
|
|
// Returns the count of replicas successfully deleted plus any error.
|
|
func cleanupOrphanSourceReplicas(ctx context.Context, clusterInfo *types.ClusterInfo, metric *types.VolumeHealthMetrics, expectedShards int) (int, error) {
|
|
if clusterInfo == nil || clusterInfo.ActiveTopology == nil {
|
|
return 0, fmt.Errorf("active topology unavailable")
|
|
}
|
|
if clusterInfo.GrpcDialOption == nil {
|
|
return 0, fmt.Errorf("grpc dial option unavailable")
|
|
}
|
|
|
|
// Re-verify shard completeness right before acting. Defensive: detection
|
|
// processes many volumes sequentially and the topology snapshot we built
|
|
// at start-of-detection could have lost shards in between (a volume
|
|
// server going down between iterations). Refusing to delete the source
|
|
// when we can no longer prove the shards are complete is the safer
|
|
// failure mode — the source replica is the only complete copy.
|
|
actualShards := countExistingEcShardsForVolume(clusterInfo.ActiveTopology, metric.VolumeID, metric.Collection)
|
|
if actualShards < expectedShards {
|
|
return 0, fmt.Errorf("EC shard set shrank between detection and cleanup (%d < %d); refusing to delete source replica", actualShards, expectedShards)
|
|
}
|
|
|
|
replicas := findVolumeReplicaLocations(clusterInfo.ActiveTopology, metric.VolumeID, metric.Collection)
|
|
if len(replicas) == 0 {
|
|
return 0, nil
|
|
}
|
|
|
|
deleted := 0
|
|
var deleteErrors []string
|
|
for _, replica := range replicas {
|
|
serverAddress := replica.ServerID
|
|
err := operation.WithVolumeServerClient(false, pb.ServerAddress(serverAddress), clusterInfo.GrpcDialOption,
|
|
func(client volume_server_pb.VolumeServerClient) error {
|
|
_, deleteErr := client.VolumeDelete(ctx, &volume_server_pb.VolumeDeleteRequest{
|
|
VolumeId: metric.VolumeID,
|
|
OnlyEmpty: false,
|
|
})
|
|
return deleteErr
|
|
})
|
|
if err != nil {
|
|
deleteErrors = append(deleteErrors, fmt.Sprintf("server %s: %v", serverAddress, err))
|
|
continue
|
|
}
|
|
deleted++
|
|
glog.V(1).Infof("EC Detection: deleted orphan regular replica for volume %d on %s", metric.VolumeID, serverAddress)
|
|
}
|
|
|
|
if len(deleteErrors) > 0 {
|
|
return deleted, fmt.Errorf("%d of %d replica delete(s) failed: %s", len(deleteErrors), len(replicas), strings.Join(deleteErrors, "; "))
|
|
}
|
|
return deleted, nil
|
|
}
|
|
|
|
// countExistingEcShardsForVolume returns the number of distinct EC shard IDs
|
|
// for (volumeID, collection) present in the topology. Walks every disk's
|
|
// EcIndexBits bitmap rather than trusting len(EcShardInfos), because a single
|
|
// info entry can carry multiple shards. Used by the #9448 guard to decide
|
|
// whether the EC shard set is complete enough that the orphaned regular
|
|
// replica is safe to delete.
|
|
func countExistingEcShardsForVolume(activeTopology *topology.ActiveTopology, volumeID uint32, collection string) int {
|
|
if activeTopology == nil {
|
|
return 0
|
|
}
|
|
topologyInfo := activeTopology.GetTopologyInfo()
|
|
if topologyInfo == nil {
|
|
return 0
|
|
}
|
|
var seen erasure_coding.ShardBits
|
|
for _, dc := range topologyInfo.DataCenterInfos {
|
|
for _, rack := range dc.RackInfos {
|
|
for _, node := range rack.DataNodeInfos {
|
|
for _, diskInfo := range node.DiskInfos {
|
|
for _, ecShardInfo := range diskInfo.EcShardInfos {
|
|
if ecShardInfo.Id != volumeID || ecShardInfo.Collection != collection {
|
|
continue
|
|
}
|
|
seen |= erasure_coding.ShardBits(ecShardInfo.EcIndexBits)
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
return seen.Count()
|
|
}
|