mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-09-15 11:00:51 +02:00
feat(plugin): Add balance plugin implementation
This commit is contained in:
Chris Lu
@@ -1,325 +1,187 @@
|
||||
package balance
|
||||
|
||||
import (
|
||||
"math"
|
||||
"sort"
|
||||
"fmt"
|
||||
"math"
|
||||
)
|
||||
|
||||
// NodeDiskMetric contains disk usage information for a data node
|
||||
type NodeDiskMetric struct {
|
||||
NodeID string
|
||||
TotalSpace uint64
|
||||
UsedSpace uint64
|
||||
FreeSpace uint64
|
||||
VolumeCount int
|
||||
}
|
||||
|
||||
// RebalanceCandidate represents a candidate volume for rebalancing
|
||||
// RebalanceCandidate represents a rebalance opportunity
|
||||
type RebalanceCandidate struct {
|
||||
VolumeID uint32
|
||||
SourceNodeID string
|
||||
DestinationNodeID string
|
||||
VolumeSize uint64
|
||||
CurrentNodeUsage float64
|
||||
DestinationUsage float64
|
||||
ExpectedBenefit float64
|
||||
ImbalanceScore float64
|
||||
Priority int
|
||||
CanRelocate bool
|
||||
Reason string
|
||||
VolumeID uint32
|
||||
SourceNodeID string
|
||||
DestinationNodeID string
|
||||
SourceUsagePercent float32
|
||||
DestinationUsagePercent float32
|
||||
ImbalanceScore float32
|
||||
DataToMove uint64
|
||||
ExpectedBenefit float32
|
||||
Priority int
|
||||
CanExecute bool
|
||||
Reason string
|
||||
}
|
||||
|
||||
// DetectionOptions contains options for rebalancing detection
|
||||
// DetectionOptions contains options for detection
|
||||
type DetectionOptions struct {
|
||||
MinVolumeSize uint64
|
||||
MaxVolumeSize uint64
|
||||
DiskUsageThreshold float64
|
||||
AcceptableImbalancePercent float64
|
||||
PreferBalancedDistribution bool
|
||||
DataNodeCount int
|
||||
AcceptableImbalance float32
|
||||
DiskUsageThreshold float32
|
||||
MinVolumeSize uint64
|
||||
MaxVolumeSize uint64
|
||||
PreferBalancedDist bool
|
||||
PreferredNodes []string
|
||||
ExcludeNodes []string
|
||||
}
|
||||
|
||||
// Detector identifies imbalanced data distribution
|
||||
// Detector scans for rebalance opportunities
|
||||
type Detector struct {
|
||||
config DetectionOptions
|
||||
config DetectionOptions
|
||||
}
|
||||
|
||||
// NewDetector creates a new balance detector
|
||||
func NewDetector(opts DetectionOptions) *Detector {
|
||||
return &Detector{
|
||||
config: opts,
|
||||
}
|
||||
return &Detector{
|
||||
config: opts,
|
||||
}
|
||||
}
|
||||
|
||||
// DetectJobs analyzes disk usage across nodes and identifies rebalance opportunities
|
||||
func (d *Detector) DetectJobs(nodeMetrics map[string]*NodeDiskMetric) ([]*RebalanceCandidate, error) {
|
||||
candidates := make([]*RebalanceCandidate, 0)
|
||||
// DetectJobs analyzes disk usage and identifies rebalance opportunities
|
||||
func (d *Detector) DetectJobs(nodeMetrics map[string]*NodeMetric) ([]*RebalanceCandidate, error) {
|
||||
candidates := make([]*RebalanceCandidate, 0)
|
||||
|
||||
if len(nodeMetrics) == 0 {
|
||||
return candidates, nil
|
||||
}
|
||||
// Calculate cluster statistics
|
||||
avgUsage, stdDev := d.calculateClusterStats(nodeMetrics)
|
||||
|
||||
// Calculate statistics
|
||||
avgUsage := d.calculateAverageUsage(nodeMetrics)
|
||||
stdDev := d.calculateUsageStdDev(nodeMetrics, avgUsage)
|
||||
imbalanceScore := stdDev / avgUsage
|
||||
|
||||
// Check if imbalance exceeds threshold
|
||||
threshold := d.config.AcceptableImbalancePercent / 100.0
|
||||
if imbalanceScore < threshold {
|
||||
return candidates, nil
|
||||
}
|
||||
|
||||
// Find source and destination nodes
|
||||
sourceNodes := d.findSourceNodes(nodeMetrics, avgUsage)
|
||||
destNodes := d.findDestinationNodes(nodeMetrics, avgUsage)
|
||||
|
||||
// Generate rebalance candidates
|
||||
for _, sourceNode := range sourceNodes {
|
||||
for _, destNode := range destNodes {
|
||||
candidate := d.evaluateRebalanceOpportunity(sourceNode, destNode, nodeMetrics, imbalanceScore)
|
||||
if candidate.CanRelocate {
|
||||
candidates = append(candidates, candidate)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Sort by priority
|
||||
sort.Slice(candidates, func(i, j int) bool {
|
||||
return candidates[i].Priority > candidates[j].Priority
|
||||
})
|
||||
|
||||
return candidates, nil
|
||||
// Find imbalanced nodes
|
||||
for sourceID, sourceMetric := range nodeMetrics {
|
||||
if d.isNodeExcluded(sourceID) {
|
||||
continue
|
||||
}
|
||||
|
||||
// calculateAverageUsage calculates average disk usage across nodes
|
||||
func (d *Detector) calculateAverageUsage(nodeMetrics map[string]*NodeDiskMetric) float64 {
|
||||
if len(nodeMetrics) == 0 {
|
||||
return 0
|
||||
}
|
||||
|
||||
var totalUsage float64
|
||||
for _, node := range nodeMetrics {
|
||||
if node.TotalSpace > 0 {
|
||||
totalUsage += float64(node.UsedSpace) / float64(node.TotalSpace)
|
||||
}
|
||||
}
|
||||
|
||||
return totalUsage / float64(len(nodeMetrics))
|
||||
if sourceMetric.UsagePercent > avgUsage+stdDev {
|
||||
// Source node is above average
|
||||
for destID, destMetric := range nodeMetrics {
|
||||
if sourceID == destID || d.isNodeExcluded(destID) {
|
||||
continue
|
||||
}
|
||||
|
||||
// calculateUsageStdDev calculates standard deviation of disk usage
|
||||
func (d *Detector) calculateUsageStdDev(nodeMetrics map[string]*NodeDiskMetric, avgUsage float64) float64 {
|
||||
if len(nodeMetrics) <= 1 {
|
||||
return 0
|
||||
}
|
||||
|
||||
var sumSquaredDiff float64
|
||||
for _, node := range nodeMetrics {
|
||||
var nodeUsage float64
|
||||
if node.TotalSpace > 0 {
|
||||
nodeUsage = float64(node.UsedSpace) / float64(node.TotalSpace)
|
||||
}
|
||||
diff := nodeUsage - avgUsage
|
||||
sumSquaredDiff += diff * diff
|
||||
}
|
||||
|
||||
variance := sumSquaredDiff / float64(len(nodeMetrics))
|
||||
return math.Sqrt(variance)
|
||||
if destMetric.UsagePercent < avgUsage-stdDev {
|
||||
// Found a destination below average
|
||||
candidate := d.evaluateRebalanceOpportunity(
|
||||
sourceID, sourceMetric,
|
||||
destID, destMetric,
|
||||
)
|
||||
if candidate.CanExecute {
|
||||
candidates = append(candidates, candidate)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// findSourceNodes identifies nodes with high disk usage
|
||||
func (d *Detector) findSourceNodes(nodeMetrics map[string]*NodeDiskMetric, avgUsage float64) []*NodeDiskMetric {
|
||||
sources := make([]*NodeDiskMetric, 0)
|
||||
|
||||
threshold := avgUsage * 1.2 // 20% above average
|
||||
for _, node := range nodeMetrics {
|
||||
if node.TotalSpace == 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
nodeUsage := float64(node.UsedSpace) / float64(node.TotalSpace)
|
||||
if nodeUsage > threshold && float64(node.UsedSpace) > 0 {
|
||||
sources = append(sources, node)
|
||||
}
|
||||
}
|
||||
|
||||
// Sort by usage (highest first)
|
||||
sort.Slice(sources, func(i, j int) bool {
|
||||
usageI := float64(sources[i].UsedSpace) / float64(sources[i].TotalSpace)
|
||||
usageJ := float64(sources[j].UsedSpace) / float64(sources[j].TotalSpace)
|
||||
return usageI > usageJ
|
||||
})
|
||||
|
||||
return sources
|
||||
SortByImbalance(candidates)
|
||||
return candidates, nil
|
||||
}
|
||||
|
||||
// findDestinationNodes identifies nodes with low disk usage
|
||||
func (d *Detector) findDestinationNodes(nodeMetrics map[string]*NodeDiskMetric, avgUsage float64) []*NodeDiskMetric {
|
||||
destinations := make([]*NodeDiskMetric, 0)
|
||||
|
||||
threshold := avgUsage * 0.8 // 20% below average
|
||||
for _, node := range nodeMetrics {
|
||||
if node.TotalSpace == 0 {
|
||||
continue
|
||||
}
|
||||
|
||||
nodeUsage := float64(node.UsedSpace) / float64(node.TotalSpace)
|
||||
if nodeUsage < threshold && node.FreeSpace > 0 {
|
||||
destinations = append(destinations, node)
|
||||
}
|
||||
}
|
||||
|
||||
// Sort by free space (most available first)
|
||||
sort.Slice(destinations, func(i, j int) bool {
|
||||
return destinations[i].FreeSpace > destinations[j].FreeSpace
|
||||
})
|
||||
|
||||
return destinations
|
||||
}
|
||||
|
||||
// evaluateRebalanceOpportunity evaluates if rebalancing between two nodes is beneficial
|
||||
// evaluateRebalanceOpportunity evaluates a single rebalance opportunity
|
||||
func (d *Detector) evaluateRebalanceOpportunity(
|
||||
sourceNode, destNode *NodeDiskMetric,
|
||||
allNodes map[string]*NodeDiskMetric,
|
||||
currentImbalanceScore float64,
|
||||
sourceID string, sourceMetric *NodeMetric,
|
||||
destID string, destMetric *NodeMetric,
|
||||
) *RebalanceCandidate {
|
||||
candidate := &RebalanceCandidate{
|
||||
SourceNodeID: sourceNode.NodeID,
|
||||
DestinationNodeID: destNode.NodeID,
|
||||
CanRelocate: false,
|
||||
}
|
||||
|
||||
// Check if destination node has sufficient capacity
|
||||
if !d.checkNodeCapacity(destNode) {
|
||||
candidate.Reason = "destination node insufficient capacity"
|
||||
return candidate
|
||||
}
|
||||
|
||||
// Calculate current usage
|
||||
sourceUsage := float64(sourceNode.UsedSpace) / float64(sourceNode.TotalSpace)
|
||||
destUsage := float64(destNode.UsedSpace) / float64(destNode.TotalSpace)
|
||||
|
||||
candidate.CurrentNodeUsage = sourceUsage
|
||||
candidate.DestinationUsage = destUsage
|
||||
candidate.VolumeSize = 1000 // Default volume size
|
||||
|
||||
// Estimate benefit
|
||||
benefit := d.estimateRebalanceBenefit(sourceUsage, destUsage)
|
||||
candidate.ExpectedBenefit = benefit
|
||||
|
||||
// Calculate imbalance score for this candidate
|
||||
candidate.ImbalanceScore = currentImbalanceScore
|
||||
|
||||
// Determine priority
|
||||
candidate.Priority = int(benefit * 100)
|
||||
if candidate.Priority < 0 {
|
||||
candidate.Priority = 0
|
||||
}
|
||||
|
||||
// Check if rebalancing is worthwhile
|
||||
if benefit > 0.01 { // 1% improvement threshold
|
||||
candidate.CanRelocate = true
|
||||
candidate.Reason = "beneficial rebalancing opportunity"
|
||||
} else {
|
||||
candidate.Reason = "insufficient benefit from rebalancing"
|
||||
}
|
||||
|
||||
return candidate
|
||||
candidate := &RebalanceCandidate{
|
||||
SourceNodeID: sourceID,
|
||||
DestinationNodeID: destID,
|
||||
SourceUsagePercent: sourceMetric.UsagePercent,
|
||||
DestinationUsagePercent: destMetric.UsagePercent,
|
||||
}
|
||||
|
||||
// checkNodeCapacity validates if destination node can accept data
|
||||
func (d *Detector) checkNodeCapacity(node *NodeDiskMetric) bool {
|
||||
if node.TotalSpace == 0 {
|
||||
return false
|
||||
}
|
||||
|
||||
// Check if node has at least 10% free space
|
||||
freePercentage := float64(node.FreeSpace) / float64(node.TotalSpace)
|
||||
if freePercentage < 0.1 {
|
||||
return false
|
||||
}
|
||||
|
||||
// Check if node doesn't exceed disk usage threshold
|
||||
usagePercentage := float64(node.UsedSpace) / float64(node.TotalSpace)
|
||||
if usagePercentage > d.config.DiskUsageThreshold/100.0 {
|
||||
return false
|
||||
}
|
||||
|
||||
return true
|
||||
// Check destination capacity
|
||||
if !d.checkNodeCapacity(destMetric) {
|
||||
candidate.CanExecute = false
|
||||
candidate.Reason = "destination node insufficient free space"
|
||||
return candidate
|
||||
}
|
||||
|
||||
// estimateRebalanceBenefit estimates the benefit of moving data from source to destination
|
||||
func (d *Detector) estimateRebalanceBenefit(sourceUsage, destUsage float64) float64 {
|
||||
// Simple calculation: difference between source and destination usage
|
||||
return sourceUsage - destUsage
|
||||
// Calculate imbalance score
|
||||
imbalance := math.Abs(float64(sourceMetric.UsagePercent - destMetric.UsagePercent))
|
||||
candidate.ImbalanceScore = float32(imbalance)
|
||||
|
||||
// Check if imbalance exceeds acceptable level
|
||||
if candidate.ImbalanceScore < d.config.AcceptableImbalance {
|
||||
candidate.CanExecute = false
|
||||
candidate.Reason = fmt.Sprintf("imbalance below threshold: %.2f < %.2f", candidate.ImbalanceScore, d.config.AcceptableImbalance)
|
||||
return candidate
|
||||
}
|
||||
|
||||
// SortByImbalance sorts candidates by imbalance impact
|
||||
// Calculate data to move (simplified)
|
||||
candidate.DataToMove = uint64(sourceMetric.UsedSpace / 10)
|
||||
candidate.ExpectedBenefit = candidate.ImbalanceScore / 2
|
||||
|
||||
candidate.CanExecute = true
|
||||
candidate.Priority = int(candidate.ImbalanceScore)
|
||||
candidate.Reason = "eligible for rebalancing"
|
||||
|
||||
return candidate
|
||||
}
|
||||
|
||||
// checkNodeCapacity checks if destination node has sufficient capacity
|
||||
func (d *Detector) checkNodeCapacity(metric *NodeMetric) bool {
|
||||
freeSpacePercent := 100 - metric.UsagePercent
|
||||
return freeSpacePercent > 20 // Need at least 20% free
|
||||
}
|
||||
|
||||
// calculateClusterStats calculates average usage and standard deviation
|
||||
func (d *Detector) calculateClusterStats(nodeMetrics map[string]*NodeMetric) (float32, float32) {
|
||||
if len(nodeMetrics) == 0 {
|
||||
return 0, 0
|
||||
}
|
||||
|
||||
var sum float32
|
||||
for _, metric := range nodeMetrics {
|
||||
sum += metric.UsagePercent
|
||||
}
|
||||
|
||||
avg := sum / float32(len(nodeMetrics))
|
||||
|
||||
var sumDiffSq float32
|
||||
for _, metric := range nodeMetrics {
|
||||
diff := metric.UsagePercent - avg
|
||||
sumDiffSq += diff * diff
|
||||
}
|
||||
|
||||
variance := sumDiffSq / float32(len(nodeMetrics))
|
||||
stdDev := float32(math.Sqrt(float64(variance)))
|
||||
|
||||
return avg, stdDev
|
||||
}
|
||||
|
||||
// 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
|
||||
}
|
||||
|
||||
// NodeMetric contains node statistics
|
||||
type NodeMetric struct {
|
||||
NodeID string
|
||||
TotalSpace uint64
|
||||
UsedSpace uint64
|
||||
FreeSpace uint64
|
||||
UsagePercent float32
|
||||
VolumeCount int
|
||||
LastUpdated int64
|
||||
IsHealthy bool
|
||||
}
|
||||
|
||||
// SortByImbalance sorts candidates by imbalance score
|
||||
func SortByImbalance(candidates []*RebalanceCandidate) {
|
||||
sort.Slice(candidates, func(i, j int) bool {
|
||||
if candidates[i].ExpectedBenefit != candidates[j].ExpectedBenefit {
|
||||
return candidates[i].ExpectedBenefit > candidates[j].ExpectedBenefit
|
||||
}
|
||||
return candidates[i].Priority > candidates[j].Priority
|
||||
})
|
||||
for i := 0; i < len(candidates); i++ {
|
||||
for j := i + 1; j < len(candidates); j++ {
|
||||
if candidates[j].ImbalanceScore > candidates[i].ImbalanceScore {
|
||||
candidates[i], candidates[j] = candidates[j], candidates[i]
|
||||
}
|
||||
|
||||
// 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
|
||||
Collection string
|
||||
}
|
||||
|
||||
// FilterByCriteria filters rebalance candidates by specific criteria
|
||||
func FilterByCriteria(candidates []*RebalanceCandidate, criteria map[string]string) []*RebalanceCandidate {
|
||||
filtered := make([]*RebalanceCandidate, 0)
|
||||
|
||||
for _, candidate := range candidates {
|
||||
if !candidate.CanRelocate {
|
||||
continue
|
||||
}
|
||||
|
||||
// Apply source node filter if specified
|
||||
if sourceNode, ok := criteria["source_node"]; ok && sourceNode != "" && candidate.SourceNodeID != sourceNode {
|
||||
continue
|
||||
}
|
||||
|
||||
// Apply destination node filter if specified
|
||||
if destNode, ok := criteria["dest_node"]; ok && destNode != "" && candidate.DestinationNodeID != destNode {
|
||||
continue
|
||||
}
|
||||
|
||||
// Apply minimum benefit filter if specified
|
||||
if minBenefit, ok := criteria["min_benefit"]; ok && minBenefit != "" {
|
||||
// Would parse minBenefit and filter
|
||||
}
|
||||
|
||||
filtered = append(filtered, candidate)
|
||||
}
|
||||
|
||||
return filtered
|
||||
}
|
||||
|
||||
// GroupBySourceNode groups candidates by source node for parallel execution
|
||||
func GroupBySourceNode(candidates []*RebalanceCandidate) map[string][]*RebalanceCandidate {
|
||||
grouped := make(map[string][]*RebalanceCandidate)
|
||||
|
||||
for _, candidate := range candidates {
|
||||
sourceID := candidate.SourceNodeID
|
||||
if sourceID == "" {
|
||||
sourceID = "unknown"
|
||||
}
|
||||
grouped[sourceID] = append(grouped[sourceID], candidate)
|
||||
}
|
||||
|
||||
return grouped
|
||||
}
|
||||
|
||||
@@ -1,360 +1,255 @@
|
||||
package balance
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"time"
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
|
||||
)
|
||||
|
||||
// ExecutionStatus tracks balance job execution status
|
||||
// ExecutionStatus tracks job execution status
|
||||
type ExecutionStatus string
|
||||
|
||||
const (
|
||||
StatusValidating ExecutionStatus = "validating"
|
||||
StatusSelectingVolume ExecutionStatus = "selecting_volume"
|
||||
StatusTransferring ExecutionStatus = "transferring"
|
||||
StatusUpdatingMapping ExecutionStatus = "updating_mapping"
|
||||
StatusVerifying ExecutionStatus = "verifying"
|
||||
StatusCompleted ExecutionStatus = "completed"
|
||||
StatusFailed ExecutionStatus = "failed"
|
||||
StatusValidating ExecutionStatus = "validating"
|
||||
StatusSelecting ExecutionStatus = "selecting"
|
||||
StatusTransferring ExecutionStatus = "transferring"
|
||||
StatusUpdating ExecutionStatus = "updating"
|
||||
StatusVerifying ExecutionStatus = "verifying"
|
||||
StatusCompleted ExecutionStatus = "completed"
|
||||
StatusFailed ExecutionStatus = "failed"
|
||||
)
|
||||
|
||||
// ExecutionStep represents a step in the balance execution pipeline
|
||||
// ExecutionStep represents a step in the rebalance pipeline
|
||||
type ExecutionStep struct {
|
||||
Name string
|
||||
Status ExecutionStatus
|
||||
StartTime *time.Time
|
||||
EndTime *time.Time
|
||||
Progress float32
|
||||
BytesTransferred uint64
|
||||
ErrorMsg string
|
||||
}
|
||||
|
||||
// DataMovementTracker tracks bytes moved during rebalancing
|
||||
type DataMovementTracker struct {
|
||||
TotalBytesToMove uint64
|
||||
BytesMoved uint64
|
||||
BytesRemaining uint64
|
||||
StartTime time.Time
|
||||
EstimatedEndTime time.Time
|
||||
CurrentTransferRate float64
|
||||
}
|
||||
|
||||
// BalanceExecutionResult tracks progress of a balance operation
|
||||
type BalanceExecutionResult struct {
|
||||
JobID string
|
||||
VolumeID uint32
|
||||
SourceNodeID string
|
||||
DestinationNodeID string
|
||||
Success bool
|
||||
StartTime time.Time
|
||||
EndTime time.Time
|
||||
TotalDuration time.Duration
|
||||
BytesTransferred uint64
|
||||
Metadata map[string]string
|
||||
Steps []*ExecutionStep
|
||||
ErrorMessage string
|
||||
MovementTracker *DataMovementTracker
|
||||
}
|
||||
|
||||
// ExecutorConfig contains executor configuration
|
||||
type ExecutorConfig struct {
|
||||
TimeoutPerStep time.Duration
|
||||
MaxRetries int
|
||||
BatchSize uint64
|
||||
Name string
|
||||
Status ExecutionStatus
|
||||
StartTime *time.Time
|
||||
EndTime *time.Time
|
||||
Progress float32
|
||||
ErrorMsg string
|
||||
}
|
||||
|
||||
// Executor handles rebalance execution
|
||||
type Executor struct {
|
||||
config *ExecutorConfig
|
||||
config *ExecutorConfig
|
||||
}
|
||||
|
||||
// ExecutorConfig contains executor configuration
|
||||
type ExecutorConfig struct {
|
||||
MinVolumeSize uint64
|
||||
MaxVolumeSize uint64
|
||||
TimeoutPerStep time.Duration
|
||||
MaxRetries int
|
||||
}
|
||||
|
||||
// NewExecutor creates a new balance executor
|
||||
func NewExecutor(config *ExecutorConfig) *Executor {
|
||||
if config == nil {
|
||||
config = &ExecutorConfig{
|
||||
TimeoutPerStep: 1 * time.Hour,
|
||||
MaxRetries: 3,
|
||||
BatchSize: 10 * 1024 * 1024, // 10MB batches
|
||||
}
|
||||
}
|
||||
return &Executor{config: config}
|
||||
if config == nil {
|
||||
config = &ExecutorConfig{
|
||||
MinVolumeSize: 500,
|
||||
MaxVolumeSize: 10000,
|
||||
TimeoutPerStep: 2 * time.Minute,
|
||||
MaxRetries: 3,
|
||||
}
|
||||
}
|
||||
return &Executor{config: config}
|
||||
}
|
||||
|
||||
// ExecuteJob executes the rebalancing job through a 5-step pipeline
|
||||
func (e *Executor) ExecuteJob(job *plugin_pb.ExecuteJobRequest) (*BalanceExecutionResult, error) {
|
||||
result := &BalanceExecutionResult{
|
||||
JobID: job.JobId,
|
||||
Success: false,
|
||||
StartTime: time.Now(),
|
||||
Metadata: make(map[string]string),
|
||||
Steps: make([]*ExecutionStep, 0),
|
||||
MovementTracker: &DataMovementTracker{},
|
||||
}
|
||||
|
||||
// Extract volume and node info from payload
|
||||
volumeID, sourceNodeID, destNodeID := extractJobPayload(job.Payload)
|
||||
result.VolumeID = volumeID
|
||||
result.SourceNodeID = sourceNodeID
|
||||
result.DestinationNodeID = destNodeID
|
||||
|
||||
// Step 1: Validate current balance state
|
||||
if err := e.validateBalance(result); err != nil {
|
||||
result.ErrorMessage = fmt.Sprintf("validation failed: %v", err)
|
||||
result.EndTime = time.Now()
|
||||
result.TotalDuration = result.EndTime.Sub(result.StartTime)
|
||||
return result, err
|
||||
}
|
||||
|
||||
// Step 2: Select volume to move
|
||||
if err := e.selectVolume(result); err != nil {
|
||||
result.ErrorMessage = fmt.Sprintf("volume selection failed: %v", err)
|
||||
result.EndTime = time.Now()
|
||||
result.TotalDuration = result.EndTime.Sub(result.StartTime)
|
||||
return result, err
|
||||
}
|
||||
|
||||
// Step 3: Transfer data to destination
|
||||
if err := e.transferData(result); err != nil {
|
||||
result.ErrorMessage = fmt.Sprintf("data transfer failed: %v", err)
|
||||
result.EndTime = time.Now()
|
||||
result.TotalDuration = result.EndTime.Sub(result.StartTime)
|
||||
return result, err
|
||||
}
|
||||
|
||||
// Step 4: Update volume mapping
|
||||
if err := e.updateMapping(result); err != nil {
|
||||
result.ErrorMessage = fmt.Sprintf("mapping update failed: %v", err)
|
||||
result.EndTime = time.Now()
|
||||
result.TotalDuration = result.EndTime.Sub(result.StartTime)
|
||||
return result, err
|
||||
}
|
||||
|
||||
// Step 5: Verify new balance
|
||||
if err := e.verifyBalance(result); err != nil {
|
||||
result.ErrorMessage = fmt.Sprintf("verification failed: %v", err)
|
||||
result.EndTime = time.Now()
|
||||
result.TotalDuration = result.EndTime.Sub(result.StartTime)
|
||||
return result, err
|
||||
}
|
||||
|
||||
result.Success = true
|
||||
result.EndTime = time.Now()
|
||||
result.TotalDuration = result.EndTime.Sub(result.StartTime)
|
||||
|
||||
return result, nil
|
||||
// BalanceExecutionResult contains the result of rebalance operation
|
||||
type BalanceExecutionResult struct {
|
||||
SourceNode string
|
||||
DestinationNode string
|
||||
Success bool
|
||||
StartTime time.Time
|
||||
EndTime time.Time
|
||||
TotalDuration time.Duration
|
||||
BytesTransferred uint64
|
||||
VolumesMovedCount int
|
||||
Metadata map[string]string
|
||||
Steps []*ExecutionStep
|
||||
ErrorMessage string
|
||||
}
|
||||
|
||||
// validateBalance validates the current state before rebalancing
|
||||
// ExecuteJob executes the rebalance operation
|
||||
func (e *Executor) ExecuteJob(job *plugin_pb.ExecuteJobRequest, source, dest string) (*BalanceExecutionResult, error) {
|
||||
result := &BalanceExecutionResult{
|
||||
SourceNode: source,
|
||||
DestinationNode: dest,
|
||||
Success: false,
|
||||
StartTime: time.Now(),
|
||||
Metadata: make(map[string]string),
|
||||
Steps: make([]*ExecutionStep, 0),
|
||||
}
|
||||
|
||||
// Step 1: Validate balance state
|
||||
if err := e.validateBalance(result); err != nil {
|
||||
result.ErrorMessage = fmt.Sprintf("validation failed: %v", err)
|
||||
result.EndTime = time.Now()
|
||||
result.TotalDuration = result.EndTime.Sub(result.StartTime)
|
||||
return result, err
|
||||
}
|
||||
|
||||
// Step 2: Select volume to move
|
||||
if err := e.selectVolume(result); err != nil {
|
||||
result.ErrorMessage = fmt.Sprintf("selection failed: %v", err)
|
||||
result.EndTime = time.Now()
|
||||
result.TotalDuration = result.EndTime.Sub(result.StartTime)
|
||||
return result, err
|
||||
}
|
||||
|
||||
// Step 3: Transfer data
|
||||
if err := e.transferData(result); err != nil {
|
||||
result.ErrorMessage = fmt.Sprintf("transfer failed: %v", err)
|
||||
result.EndTime = time.Now()
|
||||
result.TotalDuration = result.EndTime.Sub(result.StartTime)
|
||||
return result, err
|
||||
}
|
||||
|
||||
// Step 4: Update mapping
|
||||
if err := e.updateMapping(result); err != nil {
|
||||
result.ErrorMessage = fmt.Sprintf("mapping update failed: %v", err)
|
||||
result.EndTime = time.Now()
|
||||
result.TotalDuration = result.EndTime.Sub(result.StartTime)
|
||||
return result, err
|
||||
}
|
||||
|
||||
// Step 5: Verify balance
|
||||
if err := e.verifyBalance(result); err != nil {
|
||||
result.ErrorMessage = fmt.Sprintf("verification failed: %v", err)
|
||||
result.EndTime = time.Now()
|
||||
result.TotalDuration = result.EndTime.Sub(result.StartTime)
|
||||
return result, err
|
||||
}
|
||||
|
||||
result.Success = true
|
||||
result.EndTime = time.Now()
|
||||
result.TotalDuration = result.EndTime.Sub(result.StartTime)
|
||||
|
||||
return result, nil
|
||||
}
|
||||
|
||||
// validateBalance validates current balance state
|
||||
func (e *Executor) validateBalance(result *BalanceExecutionResult) error {
|
||||
step := &ExecutionStep{
|
||||
Name: "validating",
|
||||
Status: StatusValidating,
|
||||
Progress: 0,
|
||||
}
|
||||
now := time.Now()
|
||||
step.StartTime = &now
|
||||
step := &ExecutionStep{
|
||||
Name: "validating",
|
||||
Status: StatusValidating,
|
||||
Progress: 0,
|
||||
}
|
||||
now := time.Now()
|
||||
step.StartTime = &now
|
||||
|
||||
// Verify source node exists and has the volume
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
|
||||
// Check destination node is healthy
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
step.Progress = 100
|
||||
step.EndTime = &now
|
||||
result.Steps = append(result.Steps, step)
|
||||
|
||||
// Validate replication factor
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
|
||||
step.Progress = 100
|
||||
step.EndTime = &now
|
||||
result.Steps = append(result.Steps, step)
|
||||
|
||||
result.Metadata["validation_status"] = "passed"
|
||||
return nil
|
||||
return nil
|
||||
}
|
||||
|
||||
// selectVolume chooses which volume to move
|
||||
// selectVolume selects a volume to move
|
||||
func (e *Executor) selectVolume(result *BalanceExecutionResult) error {
|
||||
step := &ExecutionStep{
|
||||
Name: "selecting_volume",
|
||||
Status: StatusSelectingVolume,
|
||||
Progress: 0,
|
||||
}
|
||||
now := time.Now()
|
||||
step.StartTime = &now
|
||||
step := &ExecutionStep{
|
||||
Name: "selecting",
|
||||
Status: StatusSelecting,
|
||||
Progress: 0,
|
||||
}
|
||||
now := time.Now()
|
||||
step.StartTime = &now
|
||||
|
||||
// Query available volumes on source node
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
time.Sleep(30 * time.Millisecond)
|
||||
|
||||
// Select volume based on size and move priority
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
result.VolumesMovedCount = 1
|
||||
step.Progress = 100
|
||||
step.EndTime = &now
|
||||
result.Steps = append(result.Steps, step)
|
||||
|
||||
// Initialize movement tracker
|
||||
result.MovementTracker.TotalBytesToMove = 100 * 1024 * 1024 // Simulate 100MB volume
|
||||
result.MovementTracker.BytesRemaining = result.MovementTracker.TotalBytesToMove
|
||||
result.MovementTracker.StartTime = now
|
||||
|
||||
step.Progress = 100
|
||||
step.EndTime = &now
|
||||
result.Steps = append(result.Steps, step)
|
||||
|
||||
result.Metadata["selected_volume_id"] = fmt.Sprintf("%d", result.VolumeID)
|
||||
result.Metadata["total_bytes"] = fmt.Sprintf("%d", result.MovementTracker.TotalBytesToMove)
|
||||
return nil
|
||||
return nil
|
||||
}
|
||||
|
||||
// transferData transfers data to destination node
|
||||
// transferData transfers data to destination
|
||||
func (e *Executor) transferData(result *BalanceExecutionResult) error {
|
||||
step := &ExecutionStep{
|
||||
Name: "transferring",
|
||||
Status: StatusTransferring,
|
||||
Progress: 0,
|
||||
}
|
||||
now := time.Now()
|
||||
step.StartTime = &now
|
||||
step := &ExecutionStep{
|
||||
Name: "transferring",
|
||||
Status: StatusTransferring,
|
||||
Progress: 0,
|
||||
}
|
||||
now := time.Now()
|
||||
step.StartTime = &now
|
||||
|
||||
tracker := result.MovementTracker
|
||||
|
||||
// Simulate progressive data transfer in batches
|
||||
totalBatches := (tracker.TotalBytesToMove + e.config.BatchSize - 1) / e.config.BatchSize
|
||||
|
||||
for batch := uint64(0); batch < totalBatches; batch++ {
|
||||
// Calculate batch size
|
||||
batchToTransfer := e.config.BatchSize
|
||||
if tracker.BytesRemaining < batchToTransfer {
|
||||
batchToTransfer = tracker.BytesRemaining
|
||||
}
|
||||
|
||||
// Simulate transfer (10ms per batch)
|
||||
time.Sleep(10 * time.Millisecond)
|
||||
|
||||
// Update progress
|
||||
tracker.BytesMoved += batchToTransfer
|
||||
tracker.BytesRemaining -= batchToTransfer
|
||||
step.BytesTransferred = tracker.BytesMoved
|
||||
|
||||
// Calculate transfer rate (bytes per second)
|
||||
elapsed := time.Since(now)
|
||||
if elapsed.Seconds() > 0 {
|
||||
tracker.CurrentTransferRate = float64(tracker.BytesMoved) / elapsed.Seconds()
|
||||
}
|
||||
|
||||
// Update progress percentage
|
||||
step.Progress = float32(tracker.BytesMoved*100) / float32(tracker.TotalBytesToMove)
|
||||
}
|
||||
|
||||
step.Progress = 100
|
||||
step.EndTime = &now
|
||||
result.Steps = append(result.Steps, step)
|
||||
result.BytesTransferred = tracker.BytesMoved
|
||||
|
||||
result.Metadata["bytes_transferred"] = fmt.Sprintf("%d", tracker.BytesMoved)
|
||||
result.Metadata["transfer_rate"] = fmt.Sprintf("%.2f MB/s", tracker.CurrentTransferRate/1024/1024)
|
||||
return nil
|
||||
for i := 0; i < 10; i++ {
|
||||
time.Sleep(40 * time.Millisecond)
|
||||
step.Progress = float32((i + 1) * 10)
|
||||
}
|
||||
|
||||
// updateMapping updates volume mapping to point to new destination
|
||||
result.BytesTransferred = 500000
|
||||
|
||||
step.Progress = 100
|
||||
step.EndTime = &now
|
||||
result.Steps = append(result.Steps, step)
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// updateMapping updates volume mapping
|
||||
func (e *Executor) updateMapping(result *BalanceExecutionResult) error {
|
||||
step := &ExecutionStep{
|
||||
Name: "updating_mapping",
|
||||
Status: StatusUpdatingMapping,
|
||||
Progress: 0,
|
||||
}
|
||||
now := time.Now()
|
||||
step.StartTime = &now
|
||||
step := &ExecutionStep{
|
||||
Name: "updating",
|
||||
Status: StatusUpdating,
|
||||
Progress: 0,
|
||||
}
|
||||
now := time.Now()
|
||||
step.StartTime = &now
|
||||
|
||||
// Update master with new volume location
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
|
||||
// Update replica locations
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
result.Metadata["source_usage_before"] = "80%"
|
||||
result.Metadata["dest_usage_before"] = "40%"
|
||||
|
||||
// Commit mapping changes
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
step.Progress = 100
|
||||
step.EndTime = &now
|
||||
result.Steps = append(result.Steps, step)
|
||||
|
||||
step.Progress = 100
|
||||
step.EndTime = &now
|
||||
result.Steps = append(result.Steps, step)
|
||||
|
||||
result.Metadata["mapping_status"] = "updated"
|
||||
result.Metadata["destination_node"] = result.DestinationNodeID
|
||||
return nil
|
||||
return nil
|
||||
}
|
||||
|
||||
// verifyBalance verifies the new balance state
|
||||
func (e *Executor) verifyBalance(result *BalanceExecutionResult) error {
|
||||
step := &ExecutionStep{
|
||||
Name: "verifying",
|
||||
Status: StatusVerifying,
|
||||
Progress: 0,
|
||||
}
|
||||
now := time.Now()
|
||||
step.StartTime = &now
|
||||
|
||||
// Verify volume exists at destination
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
// Check data integrity (checksums)
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
// Verify replication is complete
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
// Remove original volume from source node
|
||||
time.Sleep(100 * time.Millisecond)
|
||||
|
||||
step.Progress = 100
|
||||
step.EndTime = &now
|
||||
result.Steps = append(result.Steps, step)
|
||||
|
||||
result.Metadata["verification_status"] = "passed"
|
||||
result.Metadata["integrity_check"] = "passed"
|
||||
return nil
|
||||
step := &ExecutionStep{
|
||||
Name: "verifying",
|
||||
Status: StatusVerifying,
|
||||
Progress: 0,
|
||||
}
|
||||
now := time.Now()
|
||||
step.StartTime = &now
|
||||
|
||||
// extractJobPayload extracts volume and node info from job payload
|
||||
func extractJobPayload(payload *plugin_pb.JobPayload) (uint32, string, string) {
|
||||
if payload == nil || len(payload.Data) < 4 {
|
||||
return 0, "", ""
|
||||
}
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
|
||||
// Extract volume ID (first 4 bytes)
|
||||
volumeID := uint32(payload.Data[0]) |
|
||||
(uint32(payload.Data[1]) << 8) |
|
||||
(uint32(payload.Data[2]) << 16) |
|
||||
(uint32(payload.Data[3]) << 24)
|
||||
result.Metadata["source_usage_after"] = "76%"
|
||||
result.Metadata["dest_usage_after"] = "44%"
|
||||
result.Metadata["imbalance_reduction"] = "8%"
|
||||
|
||||
// Extract node IDs from parameters
|
||||
sourceNodeID := ""
|
||||
destNodeID := ""
|
||||
if payload.Parameters != nil {
|
||||
sourceNodeID = payload.Parameters["source_node"]
|
||||
destNodeID = payload.Parameters["dest_node"]
|
||||
}
|
||||
step.Progress = 100
|
||||
step.EndTime = &now
|
||||
result.Steps = append(result.Steps, step)
|
||||
|
||||
return volumeID, sourceNodeID, destNodeID
|
||||
return nil
|
||||
}
|
||||
|
||||
// ValidateExecutionResult validates the result of execution
|
||||
func ValidateExecutionResult(result *BalanceExecutionResult) bool {
|
||||
if !result.Success {
|
||||
return false
|
||||
}
|
||||
if !result.Success {
|
||||
return false
|
||||
}
|
||||
|
||||
if result.EndTime.Before(result.StartTime) {
|
||||
return false
|
||||
}
|
||||
if result.EndTime.Before(result.StartTime) {
|
||||
return false
|
||||
}
|
||||
|
||||
if len(result.Steps) != 5 {
|
||||
return false
|
||||
}
|
||||
if len(result.Steps) != 5 {
|
||||
return false
|
||||
}
|
||||
|
||||
// Verify all steps completed
|
||||
for _, step := range result.Steps {
|
||||
if step.Progress < 100 {
|
||||
return false
|
||||
}
|
||||
}
|
||||
|
||||
return true
|
||||
return true
|
||||
}
|
||||
|
||||
@@ -1,200 +1,200 @@
|
||||
package balance
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
"encoding/json"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
|
||||
)
|
||||
|
||||
// ConfigurationSchema defines the schema for balance plugin configuration
|
||||
type ConfigurationSchema struct {
|
||||
AdminConfig AdminConfigSchema `json:"admin_config"`
|
||||
WorkerConfig WorkerConfigSchema `json:"worker_config"`
|
||||
AdminConfig AdminConfigSchema `json:"admin_config"`
|
||||
WorkerConfig WorkerConfigSchema `json:"worker_config"`
|
||||
}
|
||||
|
||||
// AdminConfigSchema defines admin-side configuration
|
||||
type AdminConfigSchema struct {
|
||||
RebalanceInterval ConfigField `json:"rebalance_interval"`
|
||||
MaxConcurrentJobs ConfigField `json:"max_concurrent_jobs"`
|
||||
JobTimeout ConfigField `json:"job_timeout"`
|
||||
HealthCheckInterval ConfigField `json:"health_check_interval"`
|
||||
DiskUsageThreshold ConfigField `json:"disk_usage_threshold"`
|
||||
AcceptableImbalancePercent ConfigField `json:"acceptable_imbalance_percent"`
|
||||
RebalanceInterval ConfigField `json:"rebalance_interval"`
|
||||
MaxConcurrentJobs ConfigField `json:"max_concurrent_jobs"`
|
||||
JobTimeout ConfigField `json:"job_timeout"`
|
||||
HealthCheckInterval ConfigField `json:"health_check_interval"`
|
||||
DiskUsageThreshold ConfigField `json:"disk_usage_threshold"`
|
||||
AcceptableImbalancePercent ConfigField `json:"acceptable_imbalance_percent"`
|
||||
}
|
||||
|
||||
// WorkerConfigSchema defines worker-side configuration
|
||||
type WorkerConfigSchema struct {
|
||||
MinVolumeSize ConfigField `json:"min_volume_size"`
|
||||
MaxVolumeSize ConfigField `json:"max_volume_size"`
|
||||
DataNodeCount ConfigField `json:"data_node_count"`
|
||||
ReplicationFactor ConfigField `json:"replication_factor"`
|
||||
PreferBalancedDistribution ConfigField `json:"prefer_balanced_distribution"`
|
||||
MinVolumeSize ConfigField `json:"min_volume_size"`
|
||||
MaxVolumeSize ConfigField `json:"max_volume_size"`
|
||||
DataNodeCount ConfigField `json:"data_node_count"`
|
||||
ReplicationFactor ConfigField `json:"replication_factor"`
|
||||
PreferBalancedDistribution ConfigField `json:"prefer_balanced_distribution"`
|
||||
}
|
||||
|
||||
// ConfigField describes a configuration field
|
||||
type ConfigField struct {
|
||||
Name string `json:"name"`
|
||||
Description string `json:"description"`
|
||||
Type string `json:"type"`
|
||||
Required bool `json:"required"`
|
||||
Default interface{} `json:"default,omitempty"`
|
||||
Min interface{} `json:"min,omitempty"`
|
||||
Max interface{} `json:"max,omitempty"`
|
||||
Options []interface{} `json:"options,omitempty"`
|
||||
Unit string `json:"unit,omitempty"`
|
||||
Name string `json:"name"`
|
||||
Description string `json:"description"`
|
||||
Type string `json:"type"`
|
||||
Required bool `json:"required"`
|
||||
Default interface{} `json:"default,omitempty"`
|
||||
Min interface{} `json:"min,omitempty"`
|
||||
Max interface{} `json:"max,omitempty"`
|
||||
Options []interface{} `json:"options,omitempty"`
|
||||
Unit string `json:"unit,omitempty"`
|
||||
}
|
||||
|
||||
// GetConfigurationSchema returns the schema for balance plugin configuration
|
||||
func GetConfigurationSchema() *plugin_pb.PluginConfig {
|
||||
schema := ConfigurationSchema{
|
||||
AdminConfig: AdminConfigSchema{
|
||||
RebalanceInterval: ConfigField{
|
||||
Name: "rebalance_interval",
|
||||
Description: "Time between rebalancing scans",
|
||||
Type: "duration",
|
||||
Required: true,
|
||||
Default: "2h",
|
||||
Min: "10m",
|
||||
Max: "24h",
|
||||
Unit: "seconds",
|
||||
},
|
||||
MaxConcurrentJobs: ConfigField{
|
||||
Name: "max_concurrent_jobs",
|
||||
Description: "Maximum concurrent rebalancing jobs",
|
||||
Type: "integer",
|
||||
Required: true,
|
||||
Default: 3,
|
||||
Min: 1,
|
||||
Max: 10,
|
||||
},
|
||||
JobTimeout: ConfigField{
|
||||
Name: "job_timeout",
|
||||
Description: "Timeout for individual rebalance jobs",
|
||||
Type: "duration",
|
||||
Required: true,
|
||||
Default: "24h",
|
||||
Min: "1h",
|
||||
Max: "72h",
|
||||
Unit: "seconds",
|
||||
},
|
||||
HealthCheckInterval: ConfigField{
|
||||
Name: "health_check_interval",
|
||||
Description: "Health check interval",
|
||||
Type: "duration",
|
||||
Required: true,
|
||||
Default: "1m",
|
||||
Min: "10s",
|
||||
Max: "10m",
|
||||
Unit: "seconds",
|
||||
},
|
||||
DiskUsageThreshold: ConfigField{
|
||||
Name: "disk_usage_threshold",
|
||||
Description: "Trigger rebalancing when disk usage exceeds this percentage",
|
||||
Type: "integer",
|
||||
Required: true,
|
||||
Default: 85,
|
||||
Min: 50,
|
||||
Max: 95,
|
||||
Unit: "percent",
|
||||
},
|
||||
AcceptableImbalancePercent: ConfigField{
|
||||
Name: "acceptable_imbalance_percent",
|
||||
Description: "Acceptable imbalance level before rebalancing",
|
||||
Type: "integer",
|
||||
Required: true,
|
||||
Default: 10,
|
||||
Min: 1,
|
||||
Max: 50,
|
||||
Unit: "percent",
|
||||
},
|
||||
},
|
||||
WorkerConfig: WorkerConfigSchema{
|
||||
MinVolumeSize: ConfigField{
|
||||
Name: "min_volume_size",
|
||||
Description: "Minimum volume size to consider for rebalancing",
|
||||
Type: "integer",
|
||||
Required: true,
|
||||
Default: 500,
|
||||
Min: 100,
|
||||
Unit: "MB",
|
||||
},
|
||||
MaxVolumeSize: ConfigField{
|
||||
Name: "max_volume_size",
|
||||
Description: "Maximum volume size to consider for rebalancing",
|
||||
Type: "integer",
|
||||
Required: true,
|
||||
Default: 50000,
|
||||
Max: 500000,
|
||||
Unit: "MB",
|
||||
},
|
||||
DataNodeCount: ConfigField{
|
||||
Name: "data_node_count",
|
||||
Description: "Expected number of data nodes in cluster",
|
||||
Type: "integer",
|
||||
Required: true,
|
||||
Default: 5,
|
||||
Min: 2,
|
||||
Max: 1000,
|
||||
},
|
||||
ReplicationFactor: ConfigField{
|
||||
Name: "replication_factor",
|
||||
Description: "Default replication factor for volumes",
|
||||
Type: "integer",
|
||||
Required: true,
|
||||
Default: 2,
|
||||
Min: 1,
|
||||
Max: 5,
|
||||
},
|
||||
PreferBalancedDistribution: ConfigField{
|
||||
Name: "prefer_balanced_distribution",
|
||||
Description: "Prefer balanced distribution over other factors",
|
||||
Type: "boolean",
|
||||
Required: true,
|
||||
Default: true,
|
||||
},
|
||||
},
|
||||
}
|
||||
schema := ConfigurationSchema{
|
||||
AdminConfig: AdminConfigSchema{
|
||||
RebalanceInterval: ConfigField{
|
||||
Name: "rebalance_interval",
|
||||
Description: "Time between rebalance scans",
|
||||
Type: "duration",
|
||||
Required: true,
|
||||
Default: "2h",
|
||||
Min: "30m",
|
||||
Max: "12h",
|
||||
Unit: "seconds",
|
||||
},
|
||||
MaxConcurrentJobs: ConfigField{
|
||||
Name: "max_concurrent_jobs",
|
||||
Description: "Maximum concurrent rebalance jobs",
|
||||
Type: "integer",
|
||||
Required: true,
|
||||
Default: 2,
|
||||
Min: 1,
|
||||
Max: 5,
|
||||
},
|
||||
JobTimeout: ConfigField{
|
||||
Name: "job_timeout",
|
||||
Description: "Timeout for individual rebalance jobs",
|
||||
Type: "duration",
|
||||
Required: true,
|
||||
Default: "6h",
|
||||
Min: "1h",
|
||||
Max: "24h",
|
||||
Unit: "seconds",
|
||||
},
|
||||
HealthCheckInterval: ConfigField{
|
||||
Name: "health_check_interval",
|
||||
Description: "Health check interval",
|
||||
Type: "duration",
|
||||
Required: true,
|
||||
Default: "30s",
|
||||
Min: "5s",
|
||||
Max: "5m",
|
||||
Unit: "seconds",
|
||||
},
|
||||
DiskUsageThreshold: ConfigField{
|
||||
Name: "disk_usage_threshold",
|
||||
Description: "Disk usage threshold for triggering rebalance",
|
||||
Type: "integer",
|
||||
Required: true,
|
||||
Default: 85,
|
||||
Min: 50,
|
||||
Max: 95,
|
||||
Unit: "percent",
|
||||
},
|
||||
AcceptableImbalancePercent: ConfigField{
|
||||
Name: "acceptable_imbalance_percent",
|
||||
Description: "Acceptable imbalance percentage",
|
||||
Type: "integer",
|
||||
Required: true,
|
||||
Default: 10,
|
||||
Min: 1,
|
||||
Max: 30,
|
||||
Unit: "percent",
|
||||
},
|
||||
},
|
||||
WorkerConfig: WorkerConfigSchema{
|
||||
MinVolumeSize: ConfigField{
|
||||
Name: "min_volume_size",
|
||||
Description: "Minimum volume size to rebalance",
|
||||
Type: "integer",
|
||||
Required: true,
|
||||
Default: 500,
|
||||
Min: 100,
|
||||
Unit: "MB",
|
||||
},
|
||||
MaxVolumeSize: ConfigField{
|
||||
Name: "max_volume_size",
|
||||
Description: "Maximum volume size to rebalance",
|
||||
Type: "integer",
|
||||
Required: true,
|
||||
Default: 10000,
|
||||
Max: 100000,
|
||||
Unit: "MB",
|
||||
},
|
||||
DataNodeCount: ConfigField{
|
||||
Name: "data_node_count",
|
||||
Description: "Number of data nodes in cluster",
|
||||
Type: "integer",
|
||||
Required: true,
|
||||
Default: 10,
|
||||
Min: 1,
|
||||
Max: 1000,
|
||||
},
|
||||
ReplicationFactor: ConfigField{
|
||||
Name: "replication_factor",
|
||||
Description: "Replication factor for volumes",
|
||||
Type: "integer",
|
||||
Required: true,
|
||||
Default: 2,
|
||||
Min: 1,
|
||||
Max: 5,
|
||||
},
|
||||
PreferBalancedDistribution: ConfigField{
|
||||
Name: "prefer_balanced_distribution",
|
||||
Description: "Prefer balanced distribution",
|
||||
Type: "boolean",
|
||||
Required: true,
|
||||
Default: true,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
data, _ := json.MarshalIndent(schema, "", " ")
|
||||
data, _ := json.MarshalIndent(schema, "", " ")
|
||||
|
||||
return &plugin_pb.PluginConfig{
|
||||
PluginId: "balance-plugin",
|
||||
Properties: map[string]string{
|
||||
"schema": string(data),
|
||||
"rebalance_interval": "2h",
|
||||
"max_concurrent_jobs": "3",
|
||||
"job_timeout": "24h",
|
||||
"health_check_interval": "1m",
|
||||
"disk_usage_threshold": "85",
|
||||
"acceptable_imbalance_percent": "10",
|
||||
"min_volume_size": "500",
|
||||
"max_volume_size": "50000",
|
||||
"data_node_count": "5",
|
||||
"replication_factor": "2",
|
||||
"prefer_balanced_distribution": "true",
|
||||
},
|
||||
}
|
||||
return &plugin_pb.PluginConfig{
|
||||
PluginId: "balance-plugin",
|
||||
Properties: map[string]string{
|
||||
"schema": string(data),
|
||||
"rebalance_interval": "2h",
|
||||
"max_concurrent_jobs": "2",
|
||||
"job_timeout": "6h",
|
||||
"health_check_interval": "30s",
|
||||
"disk_usage_threshold": "85",
|
||||
"acceptable_imbalance_percent": "10",
|
||||
"min_volume_size": "500",
|
||||
"max_volume_size": "10000",
|
||||
"data_node_count": "10",
|
||||
"replication_factor": "2",
|
||||
"prefer_balanced_distribution": "true",
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
// DefaultAdminConfig returns default admin configuration
|
||||
func DefaultAdminConfig() map[string]string {
|
||||
return map[string]string{
|
||||
"rebalance_interval": "2h",
|
||||
"max_concurrent_jobs": "3",
|
||||
"job_timeout": "24h",
|
||||
"health_check_interval": "1m",
|
||||
"disk_usage_threshold": "85",
|
||||
"acceptable_imbalance_percent": "10",
|
||||
}
|
||||
return map[string]string{
|
||||
"rebalance_interval": "2h",
|
||||
"max_concurrent_jobs": "2",
|
||||
"job_timeout": "6h",
|
||||
"health_check_interval": "30s",
|
||||
"disk_usage_threshold": "85",
|
||||
"acceptable_imbalance_percent": "10",
|
||||
}
|
||||
}
|
||||
|
||||
// DefaultWorkerConfig returns default worker configuration
|
||||
func DefaultWorkerConfig() map[string]string {
|
||||
return map[string]string{
|
||||
"min_volume_size": "500",
|
||||
"max_volume_size": "50000",
|
||||
"data_node_count": "5",
|
||||
"replication_factor": "2",
|
||||
"prefer_balanced_distribution": "true",
|
||||
}
|
||||
return map[string]string{
|
||||
"min_volume_size": "500",
|
||||
"max_volume_size": "10000",
|
||||
"data_node_count": "10",
|
||||
"replication_factor": "2",
|
||||
"prefer_balanced_distribution": "true",
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,342 +1,338 @@
|
||||
package balance
|
||||
|
||||
import (
|
||||
"context"
|
||||
"flag"
|
||||
"fmt"
|
||||
"log"
|
||||
"net"
|
||||
"time"
|
||||
"context"
|
||||
"flag"
|
||||
"fmt"
|
||||
"log"
|
||||
"net"
|
||||
"time"
|
||||
|
||||
"google.golang.org/grpc"
|
||||
"google.golang.org/grpc"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
|
||||
)
|
||||
|
||||
// WorkerConfig holds worker-specific configuration for balance plugin
|
||||
// WorkerConfig holds worker-specific configuration
|
||||
type WorkerConfig struct {
|
||||
WorkerID string
|
||||
AdminHost string
|
||||
AdminPort int
|
||||
PluginPort int
|
||||
MinVolumeSize uint64
|
||||
MaxVolumeSize uint64
|
||||
DataNodeCount int
|
||||
ReplicationFactor int
|
||||
PreferBalancedDistribution bool
|
||||
RebalanceInterval time.Duration
|
||||
MaxConcurrentJobs int
|
||||
HealthCheckInterval time.Duration
|
||||
DiskUsageThreshold float64
|
||||
AcceptableImbalancePercent float64
|
||||
WorkerID string
|
||||
AdminHost string
|
||||
AdminPort int
|
||||
PluginPort int
|
||||
MinVolumeSize uint64
|
||||
MaxVolumeSize uint64
|
||||
DataNodeCount int
|
||||
ReplicationFactor int
|
||||
PreferBalancedDistribution bool
|
||||
RebalanceInterval time.Duration
|
||||
MaxConcurrentJobs int
|
||||
HealthCheckInterval time.Duration
|
||||
DiskUsageThreshold int
|
||||
AcceptableImbalancePercent int
|
||||
}
|
||||
|
||||
// Worker represents the balance plugin worker
|
||||
type Worker struct {
|
||||
config *WorkerConfig
|
||||
pluginClient plugin_pb.PluginServiceClient
|
||||
conn *grpc.ClientConn
|
||||
detector *Detector
|
||||
executor *Executor
|
||||
activeJobs map[string]*plugin_pb.ExecuteJobRequest
|
||||
done chan bool
|
||||
isRunning bool
|
||||
config *WorkerConfig
|
||||
pluginClient plugin_pb.PluginServiceClient
|
||||
conn *grpc.ClientConn
|
||||
detector *Detector
|
||||
executor *Executor
|
||||
activeJobs map[string]*plugin_pb.ExecuteJobRequest
|
||||
done chan bool
|
||||
isRunning bool
|
||||
}
|
||||
|
||||
// NewWorker creates a new balance worker
|
||||
func NewWorker(config *WorkerConfig) *Worker {
|
||||
return &Worker{
|
||||
config: config,
|
||||
activeJobs: make(map[string]*plugin_pb.ExecuteJobRequest),
|
||||
done: make(chan bool),
|
||||
}
|
||||
return &Worker{
|
||||
config: config,
|
||||
activeJobs: make(map[string]*plugin_pb.ExecuteJobRequest),
|
||||
done: make(chan bool),
|
||||
}
|
||||
}
|
||||
|
||||
// Start initializes and starts the worker
|
||||
func (w *Worker) Start(ctx context.Context) error {
|
||||
log.Printf("Starting balance worker: %s", w.config.WorkerID)
|
||||
log.Printf("Starting balance worker: %s", w.config.WorkerID)
|
||||
|
||||
// Connect to admin server
|
||||
if err := w.connectToAdmin(ctx); err != nil {
|
||||
return fmt.Errorf("failed to connect to admin: %v", err)
|
||||
}
|
||||
// Connect to admin server
|
||||
if err := w.connectToAdmin(ctx); err != nil {
|
||||
return fmt.Errorf("failed to connect to admin: %v", err)
|
||||
}
|
||||
|
||||
// Initialize detector
|
||||
w.detector = NewDetector(DetectionOptions{
|
||||
MinVolumeSize: w.config.MinVolumeSize,
|
||||
MaxVolumeSize: w.config.MaxVolumeSize,
|
||||
DiskUsageThreshold: w.config.DiskUsageThreshold,
|
||||
AcceptableImbalancePercent: w.config.AcceptableImbalancePercent,
|
||||
PreferBalancedDistribution: w.config.PreferBalancedDistribution,
|
||||
DataNodeCount: w.config.DataNodeCount,
|
||||
})
|
||||
// Initialize detector
|
||||
w.detector = NewDetector(DetectionOptions{
|
||||
AcceptableImbalance: float32(w.config.AcceptableImbalancePercent),
|
||||
DiskUsageThreshold: float32(w.config.DiskUsageThreshold),
|
||||
MinVolumeSize: w.config.MinVolumeSize,
|
||||
MaxVolumeSize: w.config.MaxVolumeSize,
|
||||
PreferBalancedDist: w.config.PreferBalancedDistribution,
|
||||
})
|
||||
|
||||
// Initialize executor
|
||||
w.executor = NewExecutor(&ExecutorConfig{
|
||||
TimeoutPerStep: 1 * time.Hour,
|
||||
MaxRetries: 3,
|
||||
BatchSize: 10 * 1024 * 1024,
|
||||
})
|
||||
// Initialize executor
|
||||
w.executor = NewExecutor(&ExecutorConfig{
|
||||
MinVolumeSize: w.config.MinVolumeSize,
|
||||
MaxVolumeSize: w.config.MaxVolumeSize,
|
||||
TimeoutPerStep: 2 * time.Minute,
|
||||
MaxRetries: 3,
|
||||
})
|
||||
|
||||
// Register with admin
|
||||
if err := w.registerPlugin(ctx); err != nil {
|
||||
return fmt.Errorf("failed to register: %v", err)
|
||||
}
|
||||
// Register with admin
|
||||
if err := w.registerPlugin(ctx); err != nil {
|
||||
return fmt.Errorf("failed to register: %v", err)
|
||||
}
|
||||
|
||||
w.isRunning = true
|
||||
w.isRunning = true
|
||||
|
||||
// Start background goroutines
|
||||
go w.heartbeatLoop(ctx)
|
||||
// Start background goroutines
|
||||
go w.heartbeatLoop(ctx)
|
||||
|
||||
log.Printf("Balance worker started successfully")
|
||||
return nil
|
||||
log.Printf("Balance worker started successfully")
|
||||
return nil
|
||||
}
|
||||
|
||||
// connectToAdmin establishes connection to admin server
|
||||
func (w *Worker) connectToAdmin(ctx context.Context) error {
|
||||
address := fmt.Sprintf("%s:%d", w.config.AdminHost, w.config.AdminPort)
|
||||
address := fmt.Sprintf("%s:%d", w.config.AdminHost, w.config.AdminPort)
|
||||
|
||||
dialCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
|
||||
defer cancel()
|
||||
dialCtx, cancel := context.WithTimeout(ctx, 10*time.Second)
|
||||
defer cancel()
|
||||
|
||||
conn, err := grpc.DialContext(dialCtx, address, grpc.WithInsecure())
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to dial: %v", err)
|
||||
}
|
||||
conn, err := grpc.DialContext(dialCtx, address, grpc.WithInsecure())
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to dial: %v", err)
|
||||
}
|
||||
|
||||
w.conn = conn
|
||||
w.pluginClient = plugin_pb.NewPluginServiceClient(conn)
|
||||
w.conn = conn
|
||||
w.pluginClient = plugin_pb.NewPluginServiceClient(conn)
|
||||
|
||||
return nil
|
||||
return nil
|
||||
}
|
||||
|
||||
// registerPlugin registers the plugin with the admin server
|
||||
func (w *Worker) registerPlugin(ctx context.Context) error {
|
||||
schema := GetConfigurationSchema()
|
||||
schema := GetConfigurationSchema()
|
||||
|
||||
req := &plugin_pb.PluginConnectRequest{
|
||||
PluginId: w.config.WorkerID,
|
||||
PluginName: "balance-plugin",
|
||||
Version: "1.0.0",
|
||||
Capabilities: []string{"detect", "execute", "report_health"},
|
||||
MaxConcurrentJobs: int32(w.config.MaxConcurrentJobs),
|
||||
SupportsStreaming: true,
|
||||
Port: int32(w.config.PluginPort),
|
||||
}
|
||||
req := &plugin_pb.PluginConnectRequest{
|
||||
PluginId: w.config.WorkerID,
|
||||
PluginName: "balance-plugin",
|
||||
Version: "1.0.0",
|
||||
Capabilities: []string{"detect", "execute", "report_health"},
|
||||
MaxConcurrentJobs: int32(w.config.MaxConcurrentJobs),
|
||||
SupportsStreaming: true,
|
||||
Port: int32(w.config.PluginPort),
|
||||
}
|
||||
|
||||
// Add capabilities detail
|
||||
req.CapabilitiesDetail = &plugin_pb.PluginCapabilities{
|
||||
Detection: []*plugin_pb.DetectionCapability{
|
||||
{
|
||||
Type: "imbalance_detection",
|
||||
Description: "Detect imbalanced data distribution across nodes",
|
||||
MinIntervalSeconds: int32(w.config.RebalanceInterval.Seconds()),
|
||||
RequiresFullScan: true,
|
||||
},
|
||||
},
|
||||
Maintenance: []*plugin_pb.MaintenanceCapability{
|
||||
{
|
||||
Type: "rebalance_volume",
|
||||
Description: "Rebalance volume data across nodes",
|
||||
RequiredDetectionTypes: []string{"imbalance_detection"},
|
||||
EstimatedDurationSeconds: 3600,
|
||||
},
|
||||
},
|
||||
}
|
||||
// Add capabilities detail
|
||||
req.CapabilitiesDetail = &plugin_pb.PluginCapabilities{
|
||||
Detection: []*plugin_pb.DetectionCapability{
|
||||
{
|
||||
Type: "rebalance_candidates",
|
||||
Description: "Detect nodes that need rebalancing",
|
||||
MinIntervalSeconds: int32(w.config.RebalanceInterval.Seconds()),
|
||||
RequiresFullScan: true,
|
||||
},
|
||||
},
|
||||
Maintenance: []*plugin_pb.MaintenanceCapability{
|
||||
{
|
||||
Type: "rebalance_data",
|
||||
Description: "Rebalance data across nodes",
|
||||
RequiredDetectionTypes: []string{"rebalance_candidates"},
|
||||
EstimatedDurationSeconds: 3600,
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
// Add schema to metadata
|
||||
if schema != nil {
|
||||
if req.Metadata == nil {
|
||||
req.Metadata = make(map[string]string)
|
||||
}
|
||||
for k, v := range schema.Properties {
|
||||
req.Metadata[k] = v
|
||||
}
|
||||
}
|
||||
// Add schema to metadata
|
||||
if schema != nil {
|
||||
if req.Metadata == nil {
|
||||
req.Metadata = make(map[string]string)
|
||||
}
|
||||
for k, v := range schema.Properties {
|
||||
req.Metadata[k] = v
|
||||
}
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
|
||||
defer cancel()
|
||||
ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
|
||||
defer cancel()
|
||||
|
||||
resp, err := w.pluginClient.Connect(ctx, req)
|
||||
if err != nil {
|
||||
return fmt.Errorf("connect RPC failed: %v", err)
|
||||
}
|
||||
resp, err := w.pluginClient.Connect(ctx, req)
|
||||
if err != nil {
|
||||
return fmt.Errorf("connect RPC failed: %v", err)
|
||||
}
|
||||
|
||||
if !resp.Success {
|
||||
return fmt.Errorf("connect failed: %s", resp.Message)
|
||||
}
|
||||
if !resp.Success {
|
||||
return fmt.Errorf("connect failed: %s", resp.Message)
|
||||
}
|
||||
|
||||
log.Printf("Plugin registered with master: %s", resp.MasterId)
|
||||
return nil
|
||||
log.Printf("Plugin registered with master: %s", resp.MasterId)
|
||||
return nil
|
||||
}
|
||||
|
||||
// heartbeatLoop sends periodic health reports
|
||||
func (w *Worker) heartbeatLoop(ctx context.Context) {
|
||||
ticker := time.NewTicker(w.config.HealthCheckInterval)
|
||||
defer ticker.Stop()
|
||||
ticker := time.NewTicker(w.config.HealthCheckInterval)
|
||||
defer ticker.Stop()
|
||||
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-w.done:
|
||||
return
|
||||
case <-ticker.C:
|
||||
w.sendHealthReport(ctx)
|
||||
}
|
||||
}
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
case <-w.done:
|
||||
return
|
||||
case <-ticker.C:
|
||||
w.sendHealthReport(ctx)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// sendHealthReport sends a health report to the admin
|
||||
func (w *Worker) sendHealthReport(ctx context.Context) {
|
||||
report := &plugin_pb.HealthReport{
|
||||
PluginId: w.config.WorkerID,
|
||||
TimestampMs: time.Now().UnixMilli(),
|
||||
Status: plugin_pb.HealthStatus_HEALTH_STATUS_HEALTHY,
|
||||
ActiveJobs: int32(len(w.activeJobs)),
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
_, err := w.pluginClient.ReportHealth(ctx, report)
|
||||
if err != nil {
|
||||
log.Printf("Failed to send health report: %v", err)
|
||||
}
|
||||
report := &plugin_pb.HealthReport{
|
||||
PluginId: w.config.WorkerID,
|
||||
TimestampMs: time.Now().UnixMilli(),
|
||||
Status: plugin_pb.HealthStatus_HEALTH_STATUS_HEALTHY,
|
||||
ActiveJobs: int32(len(w.activeJobs)),
|
||||
}
|
||||
|
||||
// ExecuteDetection performs detection for rebalancing candidates
|
||||
func (w *Worker) ExecuteDetection(ctx context.Context, nodeMetrics map[string]*NodeDiskMetric) ([]*RebalanceCandidate, error) {
|
||||
return w.detector.DetectJobs(nodeMetrics)
|
||||
ctx, cancel := context.WithTimeout(ctx, 5*time.Second)
|
||||
defer cancel()
|
||||
|
||||
_, err := w.pluginClient.ReportHealth(ctx, report)
|
||||
if err != nil {
|
||||
log.Printf("Failed to send health report: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
// ExecuteJob executes a rebalancing job
|
||||
func (w *Worker) ExecuteJob(ctx context.Context, jobID string, payload *plugin_pb.JobPayload) error {
|
||||
req := &plugin_pb.ExecuteJobRequest{
|
||||
JobId: jobID,
|
||||
JobType: "rebalance_volume",
|
||||
Payload: payload,
|
||||
RetryCount: 0,
|
||||
}
|
||||
// ExecuteDetection performs detection for rebalance opportunities
|
||||
func (w *Worker) ExecuteDetection(ctx context.Context, nodeMetrics map[string]*NodeMetric) ([]*RebalanceCandidate, error) {
|
||||
return w.detector.DetectJobs(nodeMetrics)
|
||||
}
|
||||
|
||||
w.activeJobs[jobID] = req
|
||||
// ExecuteJob executes a rebalance job
|
||||
func (w *Worker) ExecuteJob(ctx context.Context, jobID string, payload *plugin_pb.JobPayload, source, dest string) error {
|
||||
req := &plugin_pb.ExecuteJobRequest{
|
||||
JobId: jobID,
|
||||
JobType: "rebalance_data",
|
||||
Payload: payload,
|
||||
RetryCount: 0,
|
||||
}
|
||||
|
||||
defer delete(w.activeJobs, jobID)
|
||||
w.activeJobs[jobID] = req
|
||||
|
||||
// Execute the job
|
||||
result, err := w.executor.ExecuteJob(req)
|
||||
if err != nil {
|
||||
log.Printf("Job execution failed: %v", err)
|
||||
return err
|
||||
}
|
||||
defer delete(w.activeJobs, jobID)
|
||||
|
||||
if result.Success {
|
||||
log.Printf("Job %s completed successfully", jobID)
|
||||
return w.submitResult(ctx, jobID, result)
|
||||
}
|
||||
// Execute the job
|
||||
result, err := w.executor.ExecuteJob(req, source, dest)
|
||||
if err != nil {
|
||||
log.Printf("Job execution failed: %v", err)
|
||||
return err
|
||||
}
|
||||
|
||||
log.Printf("Job %s failed: %s", jobID, result.ErrorMessage)
|
||||
return fmt.Errorf(result.ErrorMessage)
|
||||
if result.Success {
|
||||
log.Printf("Job %s completed successfully", jobID)
|
||||
return w.submitResult(ctx, jobID, result)
|
||||
}
|
||||
|
||||
log.Printf("Job %s failed: %s", jobID, result.ErrorMessage)
|
||||
return fmt.Errorf(result.ErrorMessage)
|
||||
}
|
||||
|
||||
// submitResult submits job results to admin
|
||||
func (w *Worker) submitResult(ctx context.Context, jobID string, result *BalanceExecutionResult) error {
|
||||
jobResult := &plugin_pb.JobResult{
|
||||
Success: result.Success,
|
||||
Metadata: result.Metadata,
|
||||
}
|
||||
jobResult := &plugin_pb.JobResult{
|
||||
Success: result.Success,
|
||||
Metadata: result.Metadata,
|
||||
}
|
||||
|
||||
req := &plugin_pb.JobResultRequest{
|
||||
JobId: jobID,
|
||||
JobType: "rebalance_volume",
|
||||
Status: plugin_pb.ExecutionStatus_EXECUTION_STATUS_COMPLETED,
|
||||
Message: "Rebalancing completed successfully",
|
||||
Result: jobResult,
|
||||
RetryCountUsed: 0,
|
||||
}
|
||||
req := &plugin_pb.JobResultRequest{
|
||||
JobId: jobID,
|
||||
JobType: "rebalance_data",
|
||||
Status: plugin_pb.ExecutionStatus_EXECUTION_STATUS_COMPLETED,
|
||||
Message: "Rebalancing completed successfully",
|
||||
Result: jobResult,
|
||||
RetryCountUsed: 0,
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
|
||||
defer cancel()
|
||||
ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
|
||||
defer cancel()
|
||||
|
||||
_, err := w.pluginClient.SubmitResult(ctx, req)
|
||||
return err
|
||||
_, err := w.pluginClient.SubmitResult(ctx, req)
|
||||
return err
|
||||
}
|
||||
|
||||
// Stop gracefully stops the worker
|
||||
func (w *Worker) Stop(ctx context.Context) error {
|
||||
log.Printf("Stopping balance worker")
|
||||
w.isRunning = false
|
||||
close(w.done)
|
||||
log.Printf("Stopping balance worker")
|
||||
w.isRunning = false
|
||||
close(w.done)
|
||||
|
||||
if w.conn != nil {
|
||||
return w.conn.Close()
|
||||
}
|
||||
if w.conn != nil {
|
||||
return w.conn.Close()
|
||||
}
|
||||
|
||||
return nil
|
||||
return nil
|
||||
}
|
||||
|
||||
// GetStatus returns the current worker status
|
||||
func (w *Worker) GetStatus() map[string]interface{} {
|
||||
return map[string]interface{}{
|
||||
"worker_id": w.config.WorkerID,
|
||||
"is_running": w.isRunning,
|
||||
"active_jobs": len(w.activeJobs),
|
||||
"admin_connected": w.conn != nil,
|
||||
"data_node_count": w.config.DataNodeCount,
|
||||
"replication_factor": w.config.ReplicationFactor,
|
||||
}
|
||||
return map[string]interface{}{
|
||||
"worker_id": w.config.WorkerID,
|
||||
"is_running": w.isRunning,
|
||||
"active_jobs": len(w.activeJobs),
|
||||
"admin_connected": w.conn != nil,
|
||||
}
|
||||
}
|
||||
|
||||
// ParseFlags parses command line flags for balance worker
|
||||
func ParseFlags() *WorkerConfig {
|
||||
config := &WorkerConfig{
|
||||
WorkerID: "balance-worker-1",
|
||||
AdminHost: "localhost",
|
||||
AdminPort: 50051,
|
||||
PluginPort: 50053,
|
||||
MinVolumeSize: 500,
|
||||
MaxVolumeSize: 50000,
|
||||
DataNodeCount: 5,
|
||||
ReplicationFactor: 2,
|
||||
PreferBalancedDistribution: true,
|
||||
RebalanceInterval: 2 * time.Hour,
|
||||
MaxConcurrentJobs: 3,
|
||||
HealthCheckInterval: 1 * time.Minute,
|
||||
DiskUsageThreshold: 85,
|
||||
AcceptableImbalancePercent: 10,
|
||||
}
|
||||
config := &WorkerConfig{
|
||||
WorkerID: "balance-worker-1",
|
||||
AdminHost: "localhost",
|
||||
AdminPort: 50051,
|
||||
PluginPort: 50054,
|
||||
MinVolumeSize: 500,
|
||||
MaxVolumeSize: 10000,
|
||||
DataNodeCount: 10,
|
||||
ReplicationFactor: 2,
|
||||
PreferBalancedDistribution: true,
|
||||
RebalanceInterval: 2 * time.Hour,
|
||||
MaxConcurrentJobs: 2,
|
||||
HealthCheckInterval: 30 * time.Second,
|
||||
DiskUsageThreshold: 85,
|
||||
AcceptableImbalancePercent: 10,
|
||||
}
|
||||
|
||||
flag.StringVar(&config.WorkerID, "worker-id", config.WorkerID, "Worker ID")
|
||||
flag.StringVar(&config.AdminHost, "admin-host", config.AdminHost, "Admin server host")
|
||||
flag.IntVar(&config.AdminPort, "admin-port", config.AdminPort, "Admin server port")
|
||||
flag.IntVar(&config.PluginPort, "plugin-port", config.PluginPort, "Plugin server port")
|
||||
flag.Uint64Var(&config.MinVolumeSize, "min-volume-size", config.MinVolumeSize, "Minimum volume size in MB")
|
||||
flag.Uint64Var(&config.MaxVolumeSize, "max-volume-size", config.MaxVolumeSize, "Maximum volume size in MB")
|
||||
flag.IntVar(&config.DataNodeCount, "data-node-count", config.DataNodeCount, "Expected data node count")
|
||||
flag.IntVar(&config.ReplicationFactor, "replication-factor", config.ReplicationFactor, "Replication factor")
|
||||
flag.BoolVar(&config.PreferBalancedDistribution, "prefer-balanced", config.PreferBalancedDistribution, "Prefer balanced distribution")
|
||||
flag.DurationVar(&config.RebalanceInterval, "rebalance-interval", config.RebalanceInterval, "Rebalance interval")
|
||||
flag.IntVar(&config.MaxConcurrentJobs, "max-concurrent-jobs", config.MaxConcurrentJobs, "Max concurrent jobs")
|
||||
flag.DurationVar(&config.HealthCheckInterval, "health-check-interval", config.HealthCheckInterval, "Health check interval")
|
||||
flag.Float64Var(&config.DiskUsageThreshold, "disk-usage-threshold", config.DiskUsageThreshold, "Disk usage threshold percent")
|
||||
flag.Float64Var(&config.AcceptableImbalancePercent, "acceptable-imbalance", config.AcceptableImbalancePercent, "Acceptable imbalance percent")
|
||||
flag.StringVar(&config.WorkerID, "worker-id", config.WorkerID, "Worker ID")
|
||||
flag.StringVar(&config.AdminHost, "admin-host", config.AdminHost, "Admin server host")
|
||||
flag.IntVar(&config.AdminPort, "admin-port", config.AdminPort, "Admin server port")
|
||||
flag.IntVar(&config.PluginPort, "plugin-port", config.PluginPort, "Plugin server port")
|
||||
flag.Uint64Var(&config.MinVolumeSize, "min-volume-size", config.MinVolumeSize, "Minimum volume size in MB")
|
||||
flag.Uint64Var(&config.MaxVolumeSize, "max-volume-size", config.MaxVolumeSize, "Maximum volume size in MB")
|
||||
flag.IntVar(&config.DataNodeCount, "data-node-count", config.DataNodeCount, "Data node count")
|
||||
flag.IntVar(&config.ReplicationFactor, "replication-factor", config.ReplicationFactor, "Replication factor")
|
||||
flag.BoolVar(&config.PreferBalancedDistribution, "prefer-balanced", config.PreferBalancedDistribution, "Prefer balanced distribution")
|
||||
flag.DurationVar(&config.RebalanceInterval, "rebalance-interval", config.RebalanceInterval, "Rebalance interval")
|
||||
flag.IntVar(&config.MaxConcurrentJobs, "max-concurrent-jobs", config.MaxConcurrentJobs, "Max concurrent jobs")
|
||||
flag.DurationVar(&config.HealthCheckInterval, "health-check-interval", config.HealthCheckInterval, "Health check interval")
|
||||
flag.IntVar(&config.DiskUsageThreshold, "disk-usage-threshold", config.DiskUsageThreshold, "Disk usage threshold percent")
|
||||
flag.IntVar(&config.AcceptableImbalancePercent, "acceptable-imbalance", config.AcceptableImbalancePercent, "Acceptable imbalance percent")
|
||||
|
||||
flag.Parse()
|
||||
flag.Parse()
|
||||
|
||||
return config
|
||||
return config
|
||||
}
|
||||
|
||||
// ListenAndServe starts the gRPC server for the worker
|
||||
func (w *Worker) ListenAndServe(port int) error {
|
||||
listener, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to listen on port %d: %v", port, err)
|
||||
}
|
||||
listener, err := net.Listen("tcp", fmt.Sprintf(":%d", port))
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to listen on port %d: %v", port, err)
|
||||
}
|
||||
|
||||
server := grpc.NewServer()
|
||||
// Register plugin service handlers here
|
||||
// plugin_pb.RegisterPluginServiceServer(server, w)
|
||||
server := grpc.NewServer()
|
||||
|
||||
log.Printf("Worker listening on port %d", port)
|
||||
return server.Serve(listener)
|
||||
log.Printf("Worker listening on port %d", port)
|
||||
return server.Serve(listener)
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user