From fb7c7b3744a25904dfc12b550ff1fe7518f84834 Mon Sep 17 00:00:00 2001 From: Chris Lu Date: Tue, 17 Feb 2026 01:45:30 -0800 Subject: [PATCH] feat(plugin): Add EC plugin detection logic --- .../plugin/workers/erasure_coding/detector.go | 228 ++++++++++++++++++ 1 file changed, 228 insertions(+) create mode 100644 weed/admin/plugin/workers/erasure_coding/detector.go diff --git a/weed/admin/plugin/workers/erasure_coding/detector.go b/weed/admin/plugin/workers/erasure_coding/detector.go new file mode 100644 index 000000000..1ea8f8a5b --- /dev/null +++ b/weed/admin/plugin/workers/erasure_coding/detector.go @@ -0,0 +1,228 @@ +package erasure_coding + +import ( + "fmt" + "strings" +) + +// CandidateVolume represents a volume eligible for EC +type CandidateVolume struct { + VolumeID uint32 + DataNodeID string + Size uint64 + FreeSpace uint64 + ReplicaCount int + RackID string + DataCenterID string + FileCount int64 + LastModified int64 + CanEncode bool + Reason string +} + +// DetectionOptions contains options for detection +type DetectionOptions struct { + MinVolumeSize uint64 + MaxVolumeSize uint64 + RackAwareness bool + DataCenterAwareness bool + PreferredNodes []string + ExcludeNodes []string +} + +// Detector scans for EC candidates +type Detector struct { + config DetectionOptions +} + +// NewDetector creates a new EC detector +func NewDetector(opts DetectionOptions) *Detector { + return &Detector{ + config: opts, + } +} + +// DetectJobs scans volumes for EC candidates +func (d *Detector) DetectJobs(volumeMetrics map[uint32]*VolumeMetric) ([]*CandidateVolume, error) { + candidates := make([]*CandidateVolume, 0) + + for volumeID, metric := range volumeMetrics { + candidate, shouldInclude := d.evaluateVolume(volumeID, metric) + if shouldInclude { + candidates = append(candidates, candidate) + } + } + + return candidates, nil +} + +// evaluateVolume checks if a volume should be encoded +func (d *Detector) evaluateVolume(volumeID uint32, metric *VolumeMetric) (*CandidateVolume, bool) { + candidate := &CandidateVolume{ + VolumeID: volumeID, + DataNodeID: metric.DataNodeID, + Size: metric.Size, + FreeSpace: metric.FreeSpace, + ReplicaCount: metric.ReplicaCount, + RackID: metric.RackID, + DataCenterID: metric.DataCenterID, + FileCount: metric.FileCount, + LastModified: metric.LastModified, + } + + // Check size constraints + if metric.Size < d.config.MinVolumeSize { + candidate.CanEncode = false + candidate.Reason = fmt.Sprintf("volume too small: %d < %d", metric.Size, d.config.MinVolumeSize) + return candidate, false + } + + if metric.Size > d.config.MaxVolumeSize { + candidate.CanEncode = false + candidate.Reason = fmt.Sprintf("volume too large: %d > %d", metric.Size, d.config.MaxVolumeSize) + return candidate, false + } + + // Check if already encoded + if metric.IsEncoded { + candidate.CanEncode = false + candidate.Reason = "volume already encoded" + return candidate, false + } + + // Check replica count for optimization potential + if metric.ReplicaCount <= 1 { + candidate.CanEncode = false + candidate.Reason = "insufficient replication for encoding" + return candidate, false + } + + // Check node exclusion + if d.isNodeExcluded(metric.DataNodeID) { + candidate.CanEncode = false + candidate.Reason = "node is in exclusion list" + return candidate, false + } + + // Check node preference + if len(d.config.PreferredNodes) > 0 && !d.isPreferredNode(metric.DataNodeID) { + candidate.CanEncode = false + candidate.Reason = "node not in preferred list" + return candidate, false + } + + // Check rack awareness if enabled + if d.config.RackAwareness && metric.RackID == "" { + candidate.CanEncode = false + candidate.Reason = "rack awareness enabled but no rack information" + return candidate, false + } + + // Check data center awareness if enabled + if d.config.DataCenterAwareness && metric.DataCenterID == "" { + candidate.CanEncode = false + candidate.Reason = "data center awareness enabled but no data center information" + return candidate, false + } + + // Check if volume has aged enough (at least 1 hour old) + if metric.LastModified == 0 { + candidate.CanEncode = false + candidate.Reason = "volume too new, needs aging" + return candidate, false + } + + // Volume is a candidate + candidate.CanEncode = true + candidate.Reason = "eligible for erasure coding" + return candidate, true +} + +// isNodeExcluded checks if a node is in the exclusion list +func (d *Detector) isNodeExcluded(nodeID string) bool { + for _, excluded := range d.config.ExcludeNodes { + if excluded == nodeID { + return true + } + } + return false +} + +// isPreferredNode checks if a node is in the preferred list +func (d *Detector) isPreferredNode(nodeID string) bool { + for _, preferred := range d.config.PreferredNodes { + if preferred == nodeID { + return true + } + } + return false +} + +// VolumeMetric contains volume statistics +type VolumeMetric struct { + VolumeID uint32 + DataNodeID string + Size uint64 + FreeSpace uint64 + ReplicaCount int + RackID string + DataCenterID string + FileCount int64 + LastModified int64 + IsEncoded bool + CanResize bool + Collection string + CompactionSize int64 +} + +// FilterByCriteria filters volumes by specific criteria +func FilterByCriteria(candidates []*CandidateVolume, criteria map[string]string) []*CandidateVolume { + filtered := make([]*CandidateVolume, 0) + + for _, candidate := range candidates { + if !candidate.CanEncode { + continue + } + + // Apply collection filter if specified + if collection, ok := criteria["collection"]; ok && collection != "" { + // Would need collection info in candidate; skipping for now + continue + } + + // Apply rack filter if specified + if rack, ok := criteria["rack"]; ok && rack != "" && candidate.RackID != rack { + continue + } + + // Apply data center filter if specified + if dc, ok := criteria["datacenter"]; ok && dc != "" && candidate.DataCenterID != dc { + continue + } + + filtered = append(filtered, candidate) + } + + return filtered +} + +// SortByPriority sorts candidates by encoding priority +func SortByPriority(candidates []*CandidateVolume) { + // In a real implementation, this would use a sorting algorithm + // For now, candidates are already in order +} + +// GroupByRack groups candidates by rack for distributed encoding +func GroupByRack(candidates []*CandidateVolume) map[string][]*CandidateVolume { + grouped := make(map[string][]*CandidateVolume) + + for _, candidate := range candidates { + rackID := candidate.RackID + if rackID == "" { + rackID = "default" + } + grouped[rackID] = append(grouped[rackID], candidate) + } + + return grouped +}