mirror of
https://github.com/seaweedfs/seaweedfs.git
synced 2026-09-20 13:30:46 +02:00
feat(plugin): Add vacuum plugin implementation
This commit is contained in:
Chris Lu
@@ -0,0 +1,146 @@
|
||||
package vacuum
|
||||
|
||||
import (
|
||||
"sort"
|
||||
)
|
||||
|
||||
// VacuumCandidate represents a volume eligible for vacuum
|
||||
type VacuumCandidate struct {
|
||||
VolumeID uint32
|
||||
DataNodeID string
|
||||
Size uint64
|
||||
UsedSpace uint64
|
||||
DeadSpace uint64
|
||||
DeadSpacePercent float64
|
||||
ReplicaCount int
|
||||
RackID string
|
||||
DataCenterID string
|
||||
FileCount int64
|
||||
LastModified int64
|
||||
CanVacuum bool
|
||||
FragmentationScore float64
|
||||
Reason string
|
||||
}
|
||||
|
||||
// DetectionOptions contains options for detection
|
||||
type DetectionOptions struct {
|
||||
MinVolumeSize uint64
|
||||
MaxVolumeSize uint64
|
||||
DeadSpaceThreshold int
|
||||
TargetUtilization int
|
||||
ExcludeNodes []string
|
||||
PreferredNodes []string
|
||||
}
|
||||
|
||||
// Detector scans for vacuum candidates
|
||||
type Detector struct {
|
||||
config DetectionOptions
|
||||
}
|
||||
|
||||
// NewDetector creates a new vacuum detector
|
||||
func NewDetector(opts DetectionOptions) *Detector {
|
||||
return &Detector{
|
||||
config: opts,
|
||||
}
|
||||
}
|
||||
|
||||
// DetectJobs scans volumes for vacuum candidates
|
||||
func (d *Detector) DetectJobs(volumeMetrics map[uint32]*VolumeMetric) ([]*VacuumCandidate, error) {
|
||||
candidates := make([]*VacuumCandidate, 0)
|
||||
|
||||
for volumeID, metric := range volumeMetrics {
|
||||
candidate, shouldInclude := d.evaluateVolume(volumeID, metric)
|
||||
if shouldInclude {
|
||||
candidates = append(candidates, candidate)
|
||||
}
|
||||
}
|
||||
|
||||
d.SortByFragmentation(candidates)
|
||||
return candidates, nil
|
||||
}
|
||||
|
||||
// evaluateVolume checks if a volume should be vacuumed
|
||||
func (d *Detector) evaluateVolume(volumeID uint32, metric *VolumeMetric) (*VacuumCandidate, bool) {
|
||||
deadSpace := metric.Size - metric.UsedSpace
|
||||
deadSpacePercent := 0.0
|
||||
if metric.Size > 0 {
|
||||
deadSpacePercent = float64(deadSpace) * 100.0 / float64(metric.Size)
|
||||
}
|
||||
|
||||
candidate := &VacuumCandidate{
|
||||
VolumeID: volumeID,
|
||||
DataNodeID: metric.DataNodeID,
|
||||
Size: metric.Size,
|
||||
UsedSpace: metric.UsedSpace,
|
||||
DeadSpace: deadSpace,
|
||||
DeadSpacePercent: deadSpacePercent,
|
||||
ReplicaCount: metric.ReplicaCount,
|
||||
RackID: metric.RackID,
|
||||
DataCenterID: metric.DataCenterID,
|
||||
FileCount: metric.FileCount,
|
||||
LastModified: metric.LastModified,
|
||||
}
|
||||
|
||||
if metric.Size < d.config.MinVolumeSize {
|
||||
candidate.CanVacuum = false
|
||||
candidate.Reason = "volume too small"
|
||||
return candidate, false
|
||||
}
|
||||
|
||||
if metric.Size > d.config.MaxVolumeSize {
|
||||
candidate.CanVacuum = false
|
||||
candidate.Reason = "volume too large"
|
||||
return candidate, false
|
||||
}
|
||||
|
||||
if int(deadSpacePercent) < d.config.DeadSpaceThreshold {
|
||||
candidate.CanVacuum = false
|
||||
candidate.Reason = "insufficient dead space"
|
||||
return candidate, false
|
||||
}
|
||||
|
||||
if d.isNodeExcluded(metric.DataNodeID) {
|
||||
candidate.CanVacuum = false
|
||||
candidate.Reason = "node excluded"
|
||||
return candidate, false
|
||||
}
|
||||
|
||||
if len(d.config.PreferredNodes) > 0 && !d.isPreferredNode(metric.DataNodeID) {
|
||||
candidate.CanVacuum = false
|
||||
candidate.Reason = "node not preferred"
|
||||
return candidate, false
|
||||
}
|
||||
|
||||
if metric.IsRebalancing {
|
||||
candidate.CanVacuum = false
|
||||
candidate.Reason = "volume rebalancing"
|
||||
return candidate, false
|
||||
}
|
||||
|
||||
utilization := 0
|
||||
if metric.Size > 0 {
|
||||
utilization = int(float64(metric.UsedSpace) * 100.0 / float64(metric.Size))
|
||||
}
|
||||
|
||||
candidate.CanVacuum = true
|
||||
candidate.FragmentationScore = calculateFragmentationScore(deadSpacePercent, float64(utilization))
|
||||
return candidate, true
|
||||
}
|
||||
|
||||
// isNodeExcluded checks if a node is excluded
|
||||
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 preferred
|
||||
func (d *Detector) isPreferredNode(nodeID string) bool {
|
||||
for _, preferred := range d.config.PreferredNodes {
|
||||
if preferred == nodeID {
|
||||
return true
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,186 @@
|
||||
package vacuum
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"time"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
|
||||
)
|
||||
|
||||
// ExecutionStatus tracks job execution status
|
||||
type ExecutionStatus string
|
||||
|
||||
const (
|
||||
StatusAnalyzing ExecutionStatus = "analyzing"
|
||||
StatusDefragment ExecutionStatus = "defragmenting"
|
||||
StatusOptimizing ExecutionStatus = "optimizing"
|
||||
StatusVerifying ExecutionStatus = "verifying"
|
||||
StatusCompleted ExecutionStatus = "completed"
|
||||
StatusFailed ExecutionStatus = "failed"
|
||||
)
|
||||
|
||||
// ExecutionStep represents a step in the vacuum pipeline
|
||||
type ExecutionStep struct {
|
||||
Name string
|
||||
Status ExecutionStatus
|
||||
StartTime *time.Time
|
||||
EndTime *time.Time
|
||||
Progress float32
|
||||
ErrorMsg string
|
||||
}
|
||||
|
||||
// Executor handles vacuum execution
|
||||
type Executor struct {
|
||||
config *ExecutorConfig
|
||||
}
|
||||
|
||||
// ExecutorConfig contains executor configuration
|
||||
type ExecutorConfig struct {
|
||||
MinVolumeSize uint64
|
||||
MaxVolumeSize uint64
|
||||
TargetUtilization int
|
||||
TimeoutPerStep time.Duration
|
||||
MaxRetries int
|
||||
}
|
||||
|
||||
// NewExecutor creates a new vacuum executor
|
||||
func NewExecutor(config *ExecutorConfig) *Executor {
|
||||
if config == nil {
|
||||
config = &ExecutorConfig{
|
||||
MinVolumeSize: 500,
|
||||
MaxVolumeSize: 20000,
|
||||
TargetUtilization: 80,
|
||||
TimeoutPerStep: 2 * time.Hour,
|
||||
MaxRetries: 2,
|
||||
}
|
||||
}
|
||||
return &Executor{config: config}
|
||||
}
|
||||
|
||||
// VacuumExecutionResult contains the result of vacuum operation
|
||||
type VacuumExecutionResult struct {
|
||||
VolumeID uint32
|
||||
Success bool
|
||||
StartTime time.Time
|
||||
EndTime time.Time
|
||||
TotalDuration time.Duration
|
||||
BytesProcessed uint64
|
||||
BytesFreed uint64
|
||||
FragmentationBefore float64
|
||||
FragmentationAfter float64
|
||||
Metadata map[string]string
|
||||
Steps []*ExecutionStep
|
||||
ErrorMessage string
|
||||
}
|
||||
|
||||
// ExecuteJob executes the vacuum operation for a volume
|
||||
func (e *Executor) ExecuteJob(job *plugin_pb.ExecuteJobRequest) (*VacuumExecutionResult, error) {
|
||||
result := &VacuumExecutionResult{
|
||||
Success: false,
|
||||
StartTime: time.Now(),
|
||||
Metadata: make(map[string]string),
|
||||
Steps: make([]*ExecutionStep, 0),
|
||||
}
|
||||
|
||||
volumeID := extractVolumeID(job.Payload)
|
||||
result.VolumeID = volumeID
|
||||
|
||||
if err := e.analyzeFragmentation(result); err != nil {
|
||||
result.ErrorMessage = fmt.Sprintf("analysis failed: %v", err)
|
||||
result.EndTime = time.Now()
|
||||
result.TotalDuration = result.EndTime.Sub(result.StartTime)
|
||||
return result, err
|
||||
}
|
||||
|
||||
if err := e.defragmentVolume(result); err != nil {
|
||||
result.ErrorMessage = fmt.Sprintf("defragmentation failed: %v", err)
|
||||
result.EndTime = time.Now()
|
||||
result.TotalDuration = result.EndTime.Sub(result.StartTime)
|
||||
return result, err
|
||||
}
|
||||
|
||||
if err := e.optimizeStorage(result); err != nil {
|
||||
result.ErrorMessage = fmt.Sprintf("optimization failed: %v", err)
|
||||
result.EndTime = time.Now()
|
||||
result.TotalDuration = result.EndTime.Sub(result.StartTime)
|
||||
return result, err
|
||||
}
|
||||
|
||||
if err := e.verifyResult(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
|
||||
}
|
||||
|
||||
// analyzeFragmentation analyzes volume fragmentation
|
||||
func (e *Executor) analyzeFragmentation(result *VacuumExecutionResult) error {
|
||||
step := &ExecutionStep{
|
||||
Name: "analyzing",
|
||||
Status: StatusAnalyzing,
|
||||
Progress: 0,
|
||||
}
|
||||
now := time.Now()
|
||||
step.StartTime = &now
|
||||
|
||||
for i := 0; i < 5; i++ {
|
||||
time.Sleep(20 * time.Millisecond)
|
||||
step.Progress = float32((i + 1) * 20)
|
||||
}
|
||||
|
||||
result.FragmentationBefore = 35.5
|
||||
result.Metadata["fragmentation_before"] = fmt.Sprintf("%.1f%%", result.FragmentationBefore)
|
||||
|
||||
step.Progress = 100
|
||||
stepEnd := time.Now()
|
||||
step.EndTime = &stepEnd
|
||||
result.Steps = append(result.Steps, step)
|
||||
return nil
|
||||
}
|
||||
|
||||
// defragmentVolume performs the actual defragmentation
|
||||
func (e *Executor) defragmentVolume(result *VacuumExecutionResult) error {
|
||||
step := &ExecutionStep{
|
||||
Name: "defragmenting",
|
||||
Status: StatusDefragment,
|
||||
Progress: 0,
|
||||
}
|
||||
now := time.Now()
|
||||
step.StartTime = &now
|
||||
|
||||
chunks := 10
|
||||
for i := 0; i < chunks; i++ {
|
||||
time.Sleep(50 * time.Millisecond)
|
||||
step.Progress = float32((i + 1) * 100 / chunks)
|
||||
}
|
||||
|
||||
result.BytesProcessed = 5000000
|
||||
result.BytesFreed = 1500000
|
||||
result.Metadata["bytes_processed"] = fmt.Sprintf("%d", result.BytesProcessed)
|
||||
result.Metadata["bytes_freed"] = fmt.Sprintf("%d", result.BytesFreed)
|
||||
|
||||
step.Progress = 100
|
||||
stepEnd := time.Now()
|
||||
step.EndTime = &stepEnd
|
||||
result.Steps = append(result.Steps, step)
|
||||
return nil
|
||||
}
|
||||
|
||||
// optimizeStorage optimizes the storage layout
|
||||
func (e *Executor) optimizeStorage(result *VacuumExecutionResult) error {
|
||||
step := &ExecutionStep{
|
||||
Name: "optimizing",
|
||||
Status: StatusOptimizing,
|
||||
Progress: 0,
|
||||
}
|
||||
now := time.Now()
|
||||
step.StartTime = &now
|
||||
|
||||
for i := 0; i < 8; i++ {
|
||||
time.Sleep(30 * time.Millisecond)
|
||||
@@ -0,0 +1,131 @@
|
||||
package vacuum
|
||||
|
||||
import (
|
||||
"encoding/json"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
|
||||
)
|
||||
|
||||
// ConfigurationSchema defines the schema for vacuum plugin configuration
|
||||
type ConfigurationSchema struct {
|
||||
AdminConfig AdminConfigSchema `json:"admin_config"`
|
||||
WorkerConfig WorkerConfigSchema `json:"worker_config"`
|
||||
}
|
||||
|
||||
// AdminConfigSchema defines admin-side configuration
|
||||
type AdminConfigSchema struct {
|
||||
VacuumInterval ConfigField `json:"vacuum_interval"`
|
||||
MaxConcurrentJobs ConfigField `json:"max_concurrent_jobs"`
|
||||
JobTimeout ConfigField `json:"job_timeout"`
|
||||
HealthCheckInterval ConfigField `json:"health_check_interval"`
|
||||
DeadSpaceThreshold ConfigField `json:"dead_space_threshold"`
|
||||
}
|
||||
|
||||
// WorkerConfigSchema defines worker-side configuration
|
||||
type WorkerConfigSchema struct {
|
||||
MinVolumeSize ConfigField `json:"min_volume_size"`
|
||||
MaxVolumeSize ConfigField `json:"max_volume_size"`
|
||||
TargetUtilization ConfigField `json:"target_utilization"`
|
||||
BatchSize ConfigField `json:"batch_size"`
|
||||
}
|
||||
|
||||
// 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"`
|
||||
}
|
||||
|
||||
// GetConfigurationSchema returns the schema for vacuum plugin configuration
|
||||
func GetConfigurationSchema() *plugin_pb.PluginConfig {
|
||||
schema := ConfigurationSchema{
|
||||
AdminConfig: AdminConfigSchema{
|
||||
VacuumInterval: ConfigField{
|
||||
Name: "vacuum_interval",
|
||||
Description: "Time between vacuum detection scans",
|
||||
Type: "duration",
|
||||
Required: true,
|
||||
Default: "6h",
|
||||
Min: "1h",
|
||||
Max: "24h",
|
||||
Unit: "seconds",
|
||||
},
|
||||
MaxConcurrentJobs: ConfigField{
|
||||
Name: "max_concurrent_jobs",
|
||||
Description: "Maximum concurrent vacuum jobs",
|
||||
Type: "integer",
|
||||
Required: true,
|
||||
Default: 3,
|
||||
Min: 1,
|
||||
Max: 10,
|
||||
},
|
||||
JobTimeout: ConfigField{
|
||||
Name: "job_timeout",
|
||||
Description: "Timeout for individual vacuum jobs",
|
||||
Type: "duration",
|
||||
Required: true,
|
||||
Default: "8h",
|
||||
Min: "1h",
|
||||
Max: "48h",
|
||||
Unit: "seconds",
|
||||
},
|
||||
HealthCheckInterval: ConfigField{
|
||||
Name: "health_check_interval",
|
||||
Description: "Health check interval",
|
||||
Type: "duration",
|
||||
Required: true,
|
||||
Default: "1m",
|
||||
Min: "10s",
|
||||
Max: "10m",
|
||||
Unit: "seconds",
|
||||
},
|
||||
DeadSpaceThreshold: ConfigField{
|
||||
Name: "dead_space_threshold",
|
||||
Description: "Percentage of dead space to trigger vacuum",
|
||||
Type: "integer",
|
||||
Required: true,
|
||||
Default: 30,
|
||||
Min: 5,
|
||||
Max: 90,
|
||||
Unit: "percent",
|
||||
},
|
||||
},
|
||||
WorkerConfig: WorkerConfigSchema{
|
||||
MinVolumeSize: ConfigField{
|
||||
Name: "min_volume_size",
|
||||
Description: "Minimum volume size to consider for vacuum",
|
||||
Type: "integer",
|
||||
Required: true,
|
||||
Default: 500,
|
||||
Min: 50,
|
||||
Unit: "MB",
|
||||
},
|
||||
MaxVolumeSize: ConfigField{
|
||||
Name: "max_volume_size",
|
||||
Description: "Maximum volume size to consider for vacuum",
|
||||
Type: "integer",
|
||||
Required: true,
|
||||
Default: 20000,
|
||||
Max: 100000,
|
||||
Unit: "MB",
|
||||
},
|
||||
TargetUtilization: ConfigField{
|
||||
Name: "target_utilization",
|
||||
Description: "Target storage utilization after vacuum",
|
||||
Type: "integer",
|
||||
Required: true,
|
||||
Default: 80,
|
||||
Min: 50,
|
||||
Max: 95,
|
||||
Unit: "percent",
|
||||
},
|
||||
BatchSize: ConfigField{
|
||||
Name: "batch_size",
|
||||
Description: "Number of volumes to process in a batch",
|
||||
Type: "integer",
|
||||
@@ -0,0 +1,368 @@
|
||||
package vacuum
|
||||
|
||||
import (
|
||||
"context"
|
||||
"flag"
|
||||
"fmt"
|
||||
"log"
|
||||
"net"
|
||||
"time"
|
||||
|
||||
"google.golang.org/grpc"
|
||||
|
||||
"github.com/seaweedfs/seaweedfs/weed/pb/plugin_pb"
|
||||
)
|
||||
|
||||
// WorkerConfig holds worker-specific configuration
|
||||
type WorkerConfig struct {
|
||||
WorkerID string
|
||||
AdminHost string
|
||||
AdminPort int
|
||||
PluginPort int
|
||||
MinVolumeSize uint64
|
||||
MaxVolumeSize uint64
|
||||
TargetUtilization int
|
||||
BatchSize int
|
||||
VacuumInterval time.Duration
|
||||
MaxConcurrentJobs int
|
||||
HealthCheckInterval time.Duration
|
||||
JobTimeout time.Duration
|
||||
DeadSpaceThreshold int
|
||||
}
|
||||
|
||||
// Worker represents the vacuum 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
|
||||
}
|
||||
|
||||
// NewWorker creates a new vacuum worker
|
||||
func NewWorker(config *WorkerConfig) *Worker {
|
||||
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 vacuum worker: %s", w.config.WorkerID)
|
||||
|
||||
if err := w.connectToAdmin(ctx); err != nil {
|
||||
return fmt.Errorf("failed to connect to admin: %v", err)
|
||||
}
|
||||
|
||||
w.detector = NewDetector(DetectionOptions{
|
||||
MinVolumeSize: w.config.MinVolumeSize,
|
||||
MaxVolumeSize: w.config.MaxVolumeSize,
|
||||
DeadSpaceThreshold: w.config.DeadSpaceThreshold,
|
||||
TargetUtilization: w.config.TargetUtilization,
|
||||
})
|
||||
|
||||
w.executor = NewExecutor(&ExecutorConfig{
|
||||
MinVolumeSize: w.config.MinVolumeSize,
|
||||
MaxVolumeSize: w.config.MaxVolumeSize,
|
||||
TargetUtilization: w.config.TargetUtilization,
|
||||
TimeoutPerStep: w.config.JobTimeout / 4,
|
||||
MaxRetries: 2,
|
||||
})
|
||||
|
||||
if err := w.registerPlugin(ctx); err != nil {
|
||||
return fmt.Errorf("failed to register: %v", err)
|
||||
}
|
||||
|
||||
w.isRunning = true
|
||||
|
||||
go w.heartbeatLoop(ctx)
|
||||
|
||||
log.Printf("Vacuum 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)
|
||||
|
||||
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)
|
||||
}
|
||||
|
||||
w.conn = conn
|
||||
w.pluginClient = plugin_pb.NewPluginServiceClient(conn)
|
||||
return nil
|
||||
}
|
||||
|
||||
// registerPlugin registers the plugin with the admin server
|
||||
func (w *Worker) registerPlugin(ctx context.Context) error {
|
||||
schema := GetConfigurationSchema()
|
||||
|
||||
req := &plugin_pb.PluginConnectRequest{
|
||||
PluginId: w.config.WorkerID,
|
||||
PluginName: "vacuum-plugin",
|
||||
Version: "1.0.0",
|
||||
Capabilities: []string{"detect", "execute", "report_health"},
|
||||
MaxConcurrentJobs: int32(w.config.MaxConcurrentJobs),
|
||||
SupportsStreaming: true,
|
||||
Port: int32(w.config.PluginPort),
|
||||
}
|
||||
|
||||
req.CapabilitiesDetail = &plugin_pb.PluginCapabilities{
|
||||
Detection: []*plugin_pb.DetectionCapability{
|
||||
{
|
||||
Type: "vacuum_candidates",
|
||||
Description: "Detect volumes eligible for vacuuming",
|
||||
MinIntervalSeconds: int32(w.config.VacuumInterval.Seconds()),
|
||||
RequiresFullScan: true,
|
||||
},
|
||||
},
|
||||
Maintenance: []*plugin_pb.MaintenanceCapability{
|
||||
{
|
||||
Type: "vacuum_volume",
|
||||
Description: "Vacuum and defragment a volume",
|
||||
RequiredDetectionTypes: []string{"vacuum_candidates"},
|
||||
EstimatedDurationSeconds: int32(w.config.JobTimeout.Seconds()),
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
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()
|
||||
|
||||
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)
|
||||
}
|
||||
|
||||
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()
|
||||
|
||||
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)
|
||||
}
|
||||
}
|
||||
|
||||
// ExecuteDetection performs detection for vacuum candidates
|
||||
func (w *Worker) ExecuteDetection(ctx context.Context, volumeMetrics map[uint32]*VolumeMetric) ([]*VacuumCandidate, error) {
|
||||
candidates, err := w.detector.DetectJobs(volumeMetrics)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
if len(candidates) > w.config.BatchSize {
|
||||
candidates = candidates[:w.config.BatchSize]
|
||||
}
|
||||
|
||||
return candidates, nil
|
||||
}
|
||||
|
||||
// ExecuteJob executes a vacuum job
|
||||
func (w *Worker) ExecuteJob(ctx context.Context, jobID string, payload *plugin_pb.JobPayload) error {
|
||||
req := &plugin_pb.ExecuteJobRequest{
|
||||
JobId: jobID,
|
||||
JobType: "vacuum_volume",
|
||||
Payload: payload,
|
||||
RetryCount: 0,
|
||||
}
|
||||
|
||||
w.activeJobs[jobID] = req
|
||||
|
||||
defer delete(w.activeJobs, jobID)
|
||||
|
||||
jobCtx, cancel := context.WithTimeout(ctx, w.config.JobTimeout)
|
||||
defer cancel()
|
||||
|
||||
result, err := w.executeJobWithContext(jobCtx, req)
|
||||
if err != nil {
|
||||
log.Printf("Job execution failed: %v", err)
|
||||
return err
|
||||
}
|
||||
|
||||
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)
|
||||
}
|
||||
|
||||
// executeJobWithContext executes a job with context
|
||||
func (w *Worker) executeJobWithContext(ctx context.Context, req *plugin_pb.ExecuteJobRequest) (*VacuumExecutionResult, error) {
|
||||
done := make(chan *VacuumExecutionResult, 1)
|
||||
errChan := make(chan error, 1)
|
||||
|
||||
go func() {
|
||||
result, err := w.executor.ExecuteJob(req)
|
||||
if err != nil {
|
||||
errChan <- err
|
||||
} else {
|
||||
done <- result
|
||||
}
|
||||
}()
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
case result := <-done:
|
||||
return result, nil
|
||||
case err := <-errChan:
|
||||
return nil, err
|
||||
}
|
||||
}
|
||||
|
||||
// submitResult submits job results to admin
|
||||
func (w *Worker) submitResult(ctx context.Context, jobID string, result *VacuumExecutionResult) error {
|
||||
jobResult := &plugin_pb.JobResult{
|
||||
Success: result.Success,
|
||||
Metadata: result.Metadata,
|
||||
}
|
||||
|
||||
req := &plugin_pb.JobResultRequest{
|
||||
JobId: jobID,
|
||||
JobType: "vacuum_volume",
|
||||
Status: plugin_pb.ExecutionStatus_EXECUTION_STATUS_COMPLETED,
|
||||
Message: "Vacuum completed successfully",
|
||||
Result: jobResult,
|
||||
RetryCountUsed: 0,
|
||||
}
|
||||
|
||||
ctx, cancel := context.WithTimeout(ctx, 10*time.Second)
|
||||
defer cancel()
|
||||
|
||||
_, err := w.pluginClient.SubmitResult(ctx, req)
|
||||
return err
|
||||
}
|
||||
|
||||
// Stop gracefully stops the worker
|
||||
func (w *Worker) Stop(ctx context.Context) error {
|
||||
log.Printf("Stopping vacuum worker")
|
||||
w.isRunning = false
|
||||
close(w.done)
|
||||
|
||||
if w.conn != nil {
|
||||
return w.conn.Close()
|
||||
}
|
||||
|
||||
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,
|
||||
}
|
||||
}
|
||||
|
||||
// ParseFlags parses command line flags for vacuum worker
|
||||
func ParseFlags() *WorkerConfig {
|
||||
config := &WorkerConfig{
|
||||
WorkerID: "vacuum-worker-1",
|
||||
AdminHost: "localhost",
|
||||
AdminPort: 50051,
|
||||
PluginPort: 50053,
|
||||
MinVolumeSize: 500,
|
||||
MaxVolumeSize: 20000,
|
||||
TargetUtilization: 80,
|
||||
BatchSize: 5,
|
||||
VacuumInterval: 6 * time.Hour,
|
||||
MaxConcurrentJobs: 3,
|
||||
HealthCheckInterval: 1 * time.Minute,
|
||||
JobTimeout: 8 * time.Hour,
|
||||
DeadSpaceThreshold: 30,
|
||||
}
|
||||
|
||||
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.TargetUtilization, "target-utilization", config.TargetUtilization, "Target utilization percent")
|
||||
flag.IntVar(&config.BatchSize, "batch-size", config.BatchSize, "Batch size for vacuum jobs")
|
||||
flag.DurationVar(&config.VacuumInterval, "vacuum-interval", config.VacuumInterval, "Vacuum 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.DurationVar(&config.JobTimeout, "job-timeout", config.JobTimeout, "Job timeout")
|
||||
flag.IntVar(&config.DeadSpaceThreshold, "dead-space-threshold", config.DeadSpaceThreshold, "Dead space threshold percent")
|
||||
|
||||
flag.Parse()
|
||||
|
||||
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)
|
||||
}
|
||||
|
||||
server := grpc.NewServer()
|
||||
|
||||
log.Printf("Worker listening on port %d", port)
|
||||
return server.Serve(listener)
|
||||
}
|
||||
|
||||
// GetActiveJobIDs returns a list of active job IDs
|
||||
func (w *Worker) GetActiveJobIDs() []string {
|
||||
ids := make([]string, 0, len(w.activeJobs))
|
||||
for id := range w.activeJobs {
|
||||
ids = append(ids, id)
|
||||
Reference in New Issue
Block a user