diff --git a/weed/admin/dash/admin_server.go b/weed/admin/dash/admin_server.go index 75b038a26..34cc067ba 100644 --- a/weed/admin/dash/admin_server.go +++ b/weed/admin/dash/admin_server.go @@ -1265,14 +1265,11 @@ func (as *AdminServer) GetMaintenanceWorkersData() (*MaintenanceWorkersData, err } // StartWorkerGrpcServer starts the worker gRPC server -func (s *AdminServer) StartWorkerGrpcServer(httpPort int) error { +func (s *AdminServer) StartWorkerGrpcServer(grpcPort int) error { if s.workerGrpcServer != nil { return fmt.Errorf("worker gRPC server is already running") } - // Calculate gRPC port (HTTP port + 10000) - grpcPort := httpPort + 10000 - s.workerGrpcServer = NewWorkerGrpcServer(s) return s.workerGrpcServer.StartWithTLS(grpcPort) } diff --git a/weed/command/admin.go b/weed/command/admin.go index 6ac42330c..c1b55f105 100644 --- a/weed/command/admin.go +++ b/weed/command/admin.go @@ -33,6 +33,7 @@ var ( type AdminOptions struct { port *int + grpcPort *int masters *string adminUser *string adminPassword *string @@ -42,6 +43,7 @@ type AdminOptions struct { func init() { cmdAdmin.Run = runAdmin // break init cycle a.port = cmdAdmin.Flag.Int("port", 23646, "admin server port") + a.grpcPort = cmdAdmin.Flag.Int("port.grpc", 0, "gRPC server port for worker connections (default: http port + 10000)") a.masters = cmdAdmin.Flag.String("masters", "localhost:9333", "comma-separated master servers") a.dataDir = cmdAdmin.Flag.String("dataDir", "", "directory to store admin configuration and data files") @@ -50,7 +52,7 @@ func init() { } var cmdAdmin = &Command{ - UsageLine: "admin -port=23646 -masters=localhost:9333 [-dataDir=/path/to/data]", + UsageLine: "admin -port=23646 -masters=localhost:9333 [-port.grpc=33646] [-dataDir=/path/to/data]", Short: "start SeaweedFS web admin interface", Long: `Start a web admin interface for SeaweedFS cluster management. @@ -63,12 +65,13 @@ var cmdAdmin = &Command{ - Maintenance operations The admin interface automatically discovers filers from the master servers. - A gRPC server for worker connections runs on HTTP port + 10000. + A gRPC server for worker connections runs on the configured gRPC port (default: HTTP port + 10000). Example Usage: weed admin -port=23646 -masters="master1:9333,master2:9333" weed admin -port=23646 -masters="localhost:9333" -dataDir="/var/lib/seaweedfs-admin" - weed admin -port=23646 -masters="localhost:9333" -dataDir="~/seaweedfs-admin" + weed admin -port=23646 -port.grpc=33646 -masters="localhost:9333" -dataDir="~/seaweedfs-admin" + weed admin -port=9900 -port.grpc=19900 -masters="localhost:9333" Data Directory: - If dataDir is specified, admin configuration and maintenance data is persisted @@ -128,6 +131,11 @@ func runAdmin(cmd *Command, args []string) bool { return false } + // Set default gRPC port if not specified + if *a.grpcPort == 0 { + *a.grpcPort = *a.port + 10000 + } + // Security warnings if *a.adminPassword == "" { fmt.Println("WARNING: Admin interface is running without authentication!") @@ -135,6 +143,7 @@ func runAdmin(cmd *Command, args []string) bool { } fmt.Printf("Starting SeaweedFS Admin Interface on port %d\n", *a.port) + fmt.Printf("Worker gRPC server will run on port %d\n", *a.grpcPort) fmt.Printf("Masters: %s\n", *a.masters) fmt.Printf("Filers will be discovered automatically from masters\n") if *a.dataDir != "" { @@ -232,7 +241,7 @@ func startAdminServer(ctx context.Context, options AdminOptions) error { } // Start worker gRPC server for worker connections - err = adminServer.StartWorkerGrpcServer(*options.port) + err = adminServer.StartWorkerGrpcServer(*options.grpcPort) if err != nil { return fmt.Errorf("failed to start worker gRPC server: %w", err) } diff --git a/weed/command/worker.go b/weed/command/worker.go index f217e57f7..c9d69b12b 100644 --- a/weed/command/worker.go +++ b/weed/command/worker.go @@ -3,6 +3,7 @@ package command import ( "os" "os/signal" + "path/filepath" "strings" "syscall" "time" @@ -21,7 +22,7 @@ import ( ) var cmdWorker = &Command{ - UsageLine: "worker -admin= [-capabilities=] [-maxConcurrent=]", + UsageLine: "worker -admin= [-capabilities=] [-maxConcurrent=] [-workingDir=]", Short: "start a maintenance worker to process cluster maintenance tasks", Long: `Start a maintenance worker that connects to an admin server to process maintenance tasks like vacuum, erasure coding, remote upload, and replication fixes. @@ -34,6 +35,7 @@ Examples: weed worker -admin=admin.example.com:23646 weed worker -admin=localhost:23646 -capabilities=vacuum,replication weed worker -admin=localhost:23646 -maxConcurrent=4 + weed worker -admin=localhost:23646 -workingDir=/tmp/worker `, } @@ -43,6 +45,7 @@ var ( workerMaxConcurrent = cmdWorker.Flag.Int("maxConcurrent", 2, "maximum number of concurrent tasks") workerHeartbeatInterval = cmdWorker.Flag.Duration("heartbeat", 30*time.Second, "heartbeat interval") workerTaskRequestInterval = cmdWorker.Flag.Duration("taskInterval", 5*time.Second, "task request interval") + workerWorkingDir = cmdWorker.Flag.String("workingDir", "", "working directory for the worker") ) func init() { @@ -67,6 +70,33 @@ func runWorker(cmd *Command, args []string) bool { return false } + // Set working directory and create task-specific subdirectories + var baseWorkingDir string + if *workerWorkingDir != "" { + glog.Infof("Setting working directory to: %s", *workerWorkingDir) + if err := os.Chdir(*workerWorkingDir); err != nil { + glog.Fatalf("Failed to change working directory: %v", err) + return false + } + wd, err := os.Getwd() + if err != nil { + glog.Fatalf("Failed to get working directory: %v", err) + return false + } + baseWorkingDir = wd + glog.Infof("Current working directory: %s", baseWorkingDir) + + // Create task-specific subdirectories + for _, capability := range capabilities { + taskDir := filepath.Join(baseWorkingDir, string(capability)) + if err := os.MkdirAll(taskDir, 0755); err != nil { + glog.Fatalf("Failed to create task directory %s: %v", taskDir, err) + return false + } + glog.Infof("Created task directory: %s", taskDir) + } + } + // Create worker configuration config := &types.WorkerConfig{ AdminServer: *workerAdminServer, @@ -74,6 +104,7 @@ func runWorker(cmd *Command, args []string) bool { MaxConcurrent: *workerMaxConcurrent, HeartbeatInterval: *workerHeartbeatInterval, TaskRequestInterval: *workerTaskRequestInterval, + BaseWorkingDir: baseWorkingDir, } // Create worker instance @@ -94,6 +125,21 @@ func runWorker(cmd *Command, args []string) bool { // Set admin client workerInstance.SetAdminClient(adminClient) + // Set working directory + if *workerWorkingDir != "" { + glog.Infof("Setting working directory to: %s", *workerWorkingDir) + if err := os.Chdir(*workerWorkingDir); err != nil { + glog.Fatalf("Failed to change working directory: %v", err) + return false + } + wd, err := os.Getwd() + if err != nil { + glog.Fatalf("Failed to get working directory: %v", err) + return false + } + glog.Infof("Current working directory: %s", wd) + } + // Start the worker err = workerInstance.Start() if err != nil { diff --git a/weed/worker/types/config_types.go b/weed/worker/types/config_types.go index 8e4113580..293ea6ade 100644 --- a/weed/worker/types/config_types.go +++ b/weed/worker/types/config_types.go @@ -12,6 +12,7 @@ type WorkerConfig struct { MaxConcurrent int `json:"max_concurrent"` HeartbeatInterval time.Duration `json:"heartbeat_interval"` TaskRequestInterval time.Duration `json:"task_request_interval"` + BaseWorkingDir string `json:"base_working_dir,omitempty"` CustomParameters map[string]interface{} `json:"custom_parameters,omitempty"` } diff --git a/weed/worker/types/task_types.go b/weed/worker/types/task_types.go index b0fdb009f..3e3edd9cf 100644 --- a/weed/worker/types/task_types.go +++ b/weed/worker/types/task_types.go @@ -60,6 +60,7 @@ type TaskParams struct { VolumeID uint32 `json:"volume_id,omitempty"` Server string `json:"server,omitempty"` Collection string `json:"collection,omitempty"` + WorkingDir string `json:"working_dir,omitempty"` Parameters map[string]interface{} `json:"parameters,omitempty"` } diff --git a/weed/worker/worker.go b/weed/worker/worker.go index 3b7899f07..b333c8775 100644 --- a/weed/worker/worker.go +++ b/weed/worker/worker.go @@ -257,11 +257,18 @@ func (w *Worker) executeTask(task *types.Task) { glog.Infof("Worker %s executing task %s: %s", w.id, task.ID, task.Type) + // Determine task-specific working directory + var taskWorkingDir string + if w.config.BaseWorkingDir != "" { + taskWorkingDir = fmt.Sprintf("%s/%s", w.config.BaseWorkingDir, string(task.Type)) + } + // Create task instance taskParams := types.TaskParams{ VolumeID: task.VolumeID, Server: task.Server, Collection: task.Collection, + WorkingDir: taskWorkingDir, Parameters: task.Parameters, }