diff --git a/coordinator/cmd/worker-agent/main.go b/coordinator/cmd/worker-agent/main.go index b7e5b4e..e3602ed 100644 --- a/coordinator/cmd/worker-agent/main.go +++ b/coordinator/cmd/worker-agent/main.go @@ -1,14 +1,24 @@ // Command worker-agent is the Go worker agent: a coordinator client that -// executes SDK workloads in a Python subprocess per claimed task. +// executes SDK workloads in a Python subprocess per claimed task. It also +// carries the local setup wizard (`worker-agent setup`) so a machine that +// installs only the worker can configure and start itself without a +// coordinator on site. package main import ( + "context" "flag" "fmt" "log/slog" + "net/http" "os" + "os/exec" + "os/signal" + "syscall" + "time" "github.com/emil28092005/SciMesh/coordinator/internal/agent" + "github.com/emil28092005/SciMesh/coordinator/internal/agent/setupui" ) // version is injected at build time (-ldflags "-X main.version=...") and @@ -16,13 +26,44 @@ import ( var version = "dev" func main() { - showVersion := flag.Bool("version", false, "print the build version and exit") - flag.Parse() + if len(os.Args) > 1 && os.Args[1] == "setup" { + os.Exit(runSetup(os.Args[2:])) + } + + fs := flag.NewFlagSet("worker-agent", flag.ExitOnError) + showVersion := fs.Bool("version", false, "print the build version and exit") + configPath := fs.String("config", "", "path to a JSON config file (SCIMESH_WORKER_CONFIG overrides the default)") + checkMode := fs.Bool("check", false, "run the preflight check (coordinator + local runtime) and exit 0/1") + checkURL := fs.String("coordinator-url", "", "coordinator URL to probe in --check mode") + _ = fs.Parse(os.Args[1:]) + + agent.Version = version + if *showVersion { fmt.Println("worker-agent " + version) return } - config, err := agent.LoadConfig() + + if *checkMode { + url := *checkURL + if url == "" { + url = os.Getenv("COORDINATOR_URL") + } + ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) + defer cancel() + if url == "" { + fmt.Println("check: no coordinator URL (pass --coordinator-url or set COORDINATOR_URL)") + os.Exit(1) + } + report := agent.RunCheck(ctx, url) + printCheck(report) + if !report.Coordinator.OK || !report.Python.OK || !report.Scimesh.OK { + os.Exit(1) + } + return + } + + config, err := loadConfig(*configPath) if err != nil { slog.Error("invalid configuration", "error", err) os.Exit(2) @@ -42,3 +83,96 @@ func main() { os.Exit(1) } } + +// loadConfig prefers a --config file; environment variables override the file +// (see agent.ConfigFile.Config). Without a file, the plain environment path is +// used exactly as before. +func loadConfig(configPath string) (*agent.Config, error) { + if configPath != "" { + return agent.LoadConfigFile(configPath) + } + envPath := os.Getenv("SCIMESH_WORKER_CONFIG") + if envPath != "" { + if _, err := os.Stat(envPath); err == nil { + return agent.LoadConfigFile(envPath) + } + } + return agent.LoadConfig() +} + +func printCheck(report agent.CheckReport) { + fmt.Printf("worker-agent %s\n", report.Agent) + line := func(item agent.CheckItem) string { + mark := "✗" + if item.OK { + mark = "✓" + } + detail := item.Detail + if item.Latency > 0 { + detail = fmt.Sprintf("%s (%d ms)", detail, item.Latency) + } + return fmt.Sprintf(" %s %s: %s", mark, item.Name, detail) + } + fmt.Println(line(report.Coordinator)) + fmt.Println(line(report.Auth)) + fmt.Println(line(report.Python)) + fmt.Println(line(report.Scimesh)) +} + +// runSetup serves the local setup wizard until interrupted. It binds the +// loopback interface only. +func runSetup(args []string) int { + fs := flag.NewFlagSet("worker-agent setup", flag.ContinueOnError) + port := fs.Int("port", 0, "listen port (default 12700)") + noOpen := fs.Bool("no-open", false, "do not open the browser automatically") + if err := fs.Parse(args); err != nil { + return 2 + } + logger := slog.New(slog.NewTextHandler(os.Stderr, nil)) + + ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM) + defer stop() + + server := setupui.New(logger, setupui.Options{ + Port: *port, + OpenBrowser: func(url string) { + if *noOpen { + return + } + openBrowser(url) + }, + }) + listener, err := server.Listen() + if err != nil { + logger.Error("setup wizard could not bind the loopback port", "err", err) + return 1 + } + url := "http://" + listener.Addr().String() + logger.Info("SciMesh worker setup wizard", "url", url, "press-ctrl-c-to-stop", true) + server.OpenBrowser(url) + // Block until the signal arrives (never returns an error that matters: a + // cancelled context is the normal exit path). + err = server.Serve(ctx, listener) + if err != nil && err != http.ErrServerClosed { + logger.Error("setup wizard stopped", "err", err) + return 1 + } + return 0 +} + +// openBrowser points the user's default browser at the wizard. Best-effort: +// a missing browser must never fail the setup flow. +func openBrowser(url string) { + for _, candidate := range [][]string{ + {"xdg-open", url}, + {"open", url}, + {"cmd", "/c", "start", url}, + } { + binary, err := exec.LookPath(candidate[0]) + if err != nil { + continue + } + _ = exec.Command(binary, candidate[1:]...).Start() + return + } +} diff --git a/coordinator/internal/agent/check.go b/coordinator/internal/agent/check.go new file mode 100644 index 0000000..6b788fd --- /dev/null +++ b/coordinator/internal/agent/check.go @@ -0,0 +1,108 @@ +package agent + +import ( + "context" + "encoding/json" + "fmt" + "net/http" + "os/exec" + "runtime" + "strings" + "time" +) + +// CheckItem is one line of the preflight report the setup wizard shows. +type CheckItem struct { + Name string `json:"name"` + OK bool `json:"ok"` + Detail string `json:"detail,omitempty"` + Latency int64 `json:"latency_ms,omitempty"` +} + +// CheckReport is the full preflight result of `worker-agent --check` and of +// the wizard's test step. +type CheckReport struct { + Coordinator CheckItem `json:"coordinator"` + Auth CheckItem `json:"auth"` + Python CheckItem `json:"python"` + Scimesh CheckItem `json:"scimesh"` + Agent string `json:"agent_version"` + CoordinatorVersion string `json:"coordinator_version,omitempty"` +} + +// checkHTTP runs one GET and reports reachability + latency, with a fallback +// detail message when the server answers without JSON. +func checkHTTP(ctx context.Context, url string, timeout time.Duration) (CheckItem, string) { + started := time.Now() + reqCtx, cancel := context.WithTimeout(ctx, timeout) + defer cancel() + req, err := http.NewRequestWithContext(reqCtx, http.MethodGet, url, nil) + if err != nil { + return CheckItem{Name: "coordinator", OK: false, Detail: "invalid URL"}, "" + } + resp, err := http.DefaultClient.Do(req) + if err != nil { + detail := err.Error() + if strings.Contains(detail, "connection refused") { + detail = "no coordinator answering at this address" + } + return CheckItem{Name: "coordinator", OK: false, Detail: detail}, "" + } + defer func() { _ = resp.Body.Close() }() + version := "" + if resp.StatusCode == http.StatusOK { + var body struct { + Status string `json:"status"` + } + if err := json.NewDecoder(resp.Body).Decode(&body); err == nil && body.Status == "ok" { + return CheckItem{Name: "coordinator", OK: true, Latency: time.Since(started).Milliseconds()}, version + } + } + return CheckItem{Name: "coordinator", OK: false, Detail: fmt.Sprintf("HTTP %d", resp.StatusCode)}, version +} + +// CheckCoordinator probes the coordinator's /health endpoint. +func CheckCoordinator(ctx context.Context, url string, timeout time.Duration) CheckReport { + report := CheckReport{Agent: Version} + item, _ := checkHTTP(ctx, strings.TrimRight(url, "/")+"/health", timeout) + report.Coordinator = item + report.Auth = CheckItem{Name: "auth", OK: true, Detail: "no token configured — will be checked at registration"} + return report +} + +// CheckEnvironment verifies the local runtime: Python present and the scimesh +// package importable. +func CheckEnvironment(ctx context.Context) CheckReport { + report := CheckReport{Agent: Version} + python, err := exec.LookPath("python3") + if err != nil { + report.Python = CheckItem{Name: "python", OK: false, Detail: "python3 not found on PATH"} + return report + } + report.Python = CheckItem{Name: "python", OK: true, Detail: python} + cmd := exec.CommandContext(ctx, python, "-c", "import scimesh; print(scimesh.__version__ if hasattr(scimesh, '__version__') else 'installed')") + out, err := cmd.Output() + if err != nil { + report.Scimesh = CheckItem{Name: "scimesh", OK: false, Detail: "install with: pip install scimesh"} + return report + } + report.Scimesh = CheckItem{Name: "scimesh", OK: true, Detail: strings.TrimSpace(string(out))} + return report +} + +// RunCheck combines the coordinator probe and the local environment probe; it +// is the body behind `worker-agent --check` and the wizard's test step. +func RunCheck(ctx context.Context, coordinatorURL string) CheckReport { + report := CheckCoordinator(ctx, coordinatorURL, 15*time.Second) + env := CheckEnvironment(ctx) + report.Python = env.Python + report.Scimesh = env.Scimesh + return report +} + +// Version is the agent build version; main injects it via -ldflags and the +// setup wizard mirrors it into the report. "dev" marks a local build. +var Version = "dev" + +// Platform is the host platform string shown on the wizard. +func Platform() string { return runtime.GOOS + "/" + runtime.GOARCH } diff --git a/coordinator/internal/agent/configfile.go b/coordinator/internal/agent/configfile.go new file mode 100644 index 0000000..57e63f7 --- /dev/null +++ b/coordinator/internal/agent/configfile.go @@ -0,0 +1,133 @@ +package agent + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + "time" +) + +// ConfigFile is the persisted worker configuration written by the setup +// wizard and read back by `worker-agent --config`. Environment variables +// still win: the file fills in what the environment left unset. +type ConfigFile struct { + CoordinatorURL string `json:"coordinator_url"` + Token string `json:"token,omitempty"` + WorkerKey string `json:"worker_key,omitempty"` + UserserviceURL string `json:"userservice_url,omitempty"` + WorkDir string `json:"work_dir"` + WorkerName string `json:"worker_name,omitempty"` + CPUCount int `json:"cpu_count"` + MemoryMB int `json:"memory_mb"` + TaskRunner []string `json:"task_runner,omitempty"` +} + +// DefaultConfigPath is where the setup wizard stores the worker's +// configuration. SCIMESH_WORKER_CONFIG overrides it. +func DefaultConfigPath() string { + if raw := os.Getenv("SCIMESH_WORKER_CONFIG"); raw != "" { + return raw + } + home, err := os.UserHomeDir() + if err != nil || home == "" { + return filepath.Join(".", ".scimesh-worker", "config.json") + } + return filepath.Join(home, ".scimesh-worker", "config.json") +} + +// LoadConfigFile reads and validates a persisted configuration. The file is +// created by the wizard with 0600 permissions, so no credential is exposed to +// other local users. +func LoadConfigFile(path string) (*Config, error) { + raw, err := os.ReadFile(path) + if err != nil { + return nil, fmt.Errorf("read config file: %w", err) + } + var file ConfigFile + if err := json.Unmarshal(raw, &file); err != nil { + return nil, fmt.Errorf("parse config file: %w", err) + } + return file.Config() +} + +// Config turns the file into the daemon configuration. Environment variables +// take precedence so operators can still override any value per-process. +func (f *ConfigFile) Config() (*Config, error) { + config := &Config{} + if env := os.Getenv("COORDINATOR_URL"); env != "" { + config.CoordinatorURL = env + } else { + config.CoordinatorURL = strings.TrimSpace(f.CoordinatorURL) + } + if config.CoordinatorURL == "" { + return nil, fmt.Errorf("coordinator_url is required") + } + if !strings.HasPrefix(config.CoordinatorURL, "http://") && !strings.HasPrefix(config.CoordinatorURL, "https://") { + return nil, fmt.Errorf("coordinator_url must be an absolute HTTP(S) URL") + } + if env := os.Getenv("WORKER_AUTH_TOKEN"); env != "" { + config.Token = env + } else { + config.Token = f.Token + } + if env := os.Getenv("WORKER_KEY"); env != "" { + config.WorkerKey = env + } else { + config.WorkerKey = f.WorkerKey + } + if env := os.Getenv("USERSERVICE_URL"); env != "" { + config.UserserviceURL = env + } else { + config.UserserviceURL = f.UserserviceURL + } + if env := os.Getenv("WORK_DIR"); env != "" { + config.WorkDir = env + } else if f.WorkDir != "" { + config.WorkDir = f.WorkDir + } else { + config.WorkDir = "./scimesh-agent-data" + } + if env := os.Getenv("WORKER_NAME"); env != "" { + config.WorkerName = env + } else { + config.WorkerName = f.WorkerName + } + config.CPUCount = f.CPUCount + if config.CPUCount < 1 { + config.CPUCount = 1 + } + config.MemoryMB = f.MemoryMB + if config.MemoryMB < 0 { + config.MemoryMB = 0 + } + if len(f.TaskRunner) > 0 { + config.TaskRunner = f.TaskRunner + } + if len(config.TaskRunner) == 0 { + config.TaskRunner = []string{"python", "-m", "scimesh.worker.task"} + } + config.PollInterval = 2 * time.Second + config.RequestTimeout = 30 * time.Second + config.Heartbeat = 15 * time.Second + config.Capabilities = DefaultCapabilities() + return config, nil +} + +// Save writes the configuration file, creating the parent directory and +// restricting permissions to the owner. +func SaveConfigFile(path string, file ConfigFile) error { + payload, err := json.MarshalIndent(file, "", " ") + if err != nil { + return err + } + payload = append(payload, '\n') + if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil { + return fmt.Errorf("create config directory: %w", err) + } + if err := os.WriteFile(path, payload, 0o600); err != nil { + return fmt.Errorf("write config file: %w", err) + } + return nil +} diff --git a/coordinator/internal/agent/setupui/server.go b/coordinator/internal/agent/setupui/server.go new file mode 100644 index 0000000..450ea64 --- /dev/null +++ b/coordinator/internal/agent/setupui/server.go @@ -0,0 +1,392 @@ +// Package setupui serves the local worker setup wizard: a small HTTP server +// bound to 127.0.0.1 that writes the worker's config file, runs preflight +// checks, and starts/stops the worker as a background process. It is part of +// the worker-agent binary so a machine that installs only a worker never needs +// a coordinator. +package setupui + +import ( + "context" + "embed" + "encoding/json" + "errors" + "fmt" + "io" + "log/slog" + "net" + "net/http" + "os" + "os/exec" + "path/filepath" + "strconv" + "strings" + "sync" + "syscall" + "time" + + "github.com/emil28092005/SciMesh/coordinator/internal/agent" +) + +//go:embed template.html +var templateFS embed.FS + +const ( + defaultPort = 12700 + pidFileName = "worker.pid" + logFileName = "worker.log" +) + +// Supervisor starts and stops the worker process and tracks its pid. It is an +// interface so tests can substitute a fake. +type Supervisor interface { + // Start launches `worker-agent --config ` detached, writing output + // into the log file. Returns the child pid. + Start(configPath, logPath string) (int, error) + // Stop terminates the process recorded in the pid file. + Stop() error + // Pid returns the recorded child pid, or 0 when none is recorded. + Pid() int + // Alive reports whether the recorded child is still running. + Alive() bool +} + +// PIDSupervisor is the real Supervisor: it spawns the running binary with +// --config and manages its pid file. +type PIDSupervisor struct { + mu sync.Mutex + pidPath string +} + +func NewPIDSupervisor(pidPath string) *PIDSupervisor { return &PIDSupervisor{pidPath: pidPath} } + +func (s *PIDSupervisor) Pid() int { + s.mu.Lock() + defer s.mu.Unlock() + return s.readPid() +} + +func (s *PIDSupervisor) readPid() int { + raw, err := os.ReadFile(s.pidPath) + if err != nil { + return 0 + } + pid, err := strconv.Atoi(strings.TrimSpace(string(raw))) + if err != nil || pid < 1 { + return 0 + } + return pid +} + +func (s *PIDSupervisor) Alive() bool { + pid := s.Pid() + if pid == 0 { + return false + } + // Signal 0 probes liveness without sending anything. + return syscall.Kill(pid, 0) == nil +} + +func (s *PIDSupervisor) processAlive(pid int) bool { + return syscall.Kill(pid, 0) == nil +} + +func (s *PIDSupervisor) Start(configPath, logPath string) (int, error) { + s.mu.Lock() + defer s.mu.Unlock() + if pid := s.readPid(); pid > 0 { + if s.processAlive(pid) { + return pid, fmt.Errorf("worker is already running (pid %d)", pid) + } + } + exe, err := os.Executable() + if err != nil { + return 0, fmt.Errorf("resolve worker binary: %w", err) + } + logFile, err := os.OpenFile(logPath, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o600) + if err != nil { + return 0, fmt.Errorf("open worker log: %w", err) + } + defer func() { _ = logFile.Close() }() + cmd := exec.Command(exe, "--config", configPath) + cmd.Stdout = logFile + cmd.Stderr = logFile + cmd.Stdin = nil + if err := cmd.Start(); err != nil { + return 0, fmt.Errorf("start worker: %w", err) + } + // The child inherits our stdout/stderr descriptors pointing at the log + // file, so we can close our copy; the child keeps it open. + _ = logFile.Close() + if err := os.WriteFile(s.pidPath, []byte(strconv.Itoa(cmd.Process.Pid)+"\n"), 0o600); err != nil { + _ = cmd.Process.Kill() + return 0, fmt.Errorf("write pid file: %w", err) + } + return cmd.Process.Pid, nil +} + +func (s *PIDSupervisor) Stop() error { + s.mu.Lock() + defer s.mu.Unlock() + pid := s.readPid() + if pid == 0 { + return nil + } + proc, err := os.FindProcess(pid) + if err != nil { + _ = os.Remove(s.pidPath) + return nil + } + if err := proc.Signal(os.Interrupt); err != nil { + _ = os.Remove(s.pidPath) + return nil + } + // Give the agent a moment to exit cleanly, then reap. + deadline := time.Now().Add(3 * time.Second) + for time.Now().Before(deadline) { + if !s.processAlive(pid) { + break + } + time.Sleep(100 * time.Millisecond) + } + _ = os.Remove(s.pidPath) + return nil +} + +// Server is the wizard HTTP server, bound to the loopback interface only. +type Server struct { + log *slog.Logger + cfgPath string + logPath string + dir string + sup Supervisor + openBrowser func(string) + port int +} + +// Options customises the wizard for tests and embedding. +type Options struct { + Port int + ConfigPath string + OpenBrowser func(url string) + Supervisor Supervisor + Dir string // directory for pid/log files; defaults to the config dir +} + +func New(log *slog.Logger, opts Options) *Server { + cfgPath := opts.ConfigPath + if cfgPath == "" { + cfgPath = agent.DefaultConfigPath() + } + dir := opts.Dir + if dir == "" { + dir = filepath.Dir(cfgPath) + } + sup := opts.Supervisor + if sup == nil { + sup = NewPIDSupervisor(filepath.Join(dir, pidFileName)) + } + open := opts.OpenBrowser + if open == nil { + open = func(string) {} + } + port := opts.Port + if port == 0 { + port = defaultPort + } + return &Server{log: log, cfgPath: cfgPath, logPath: filepath.Join(dir, logFileName), dir: dir, sup: sup, openBrowser: open, port: port} +} + +// Listen binds the loopback listener and returns it; Serve runs the server on +// it. Split so tests can inspect the actual ephemeral port. +func (s *Server) Listen() (net.Listener, error) { + return net.Listen("tcp", fmt.Sprintf("127.0.0.1:%d", s.port)) +} + +// OpenBrowser hands the wizard URL to the configured opener (default: no-op). +func (s *Server) OpenBrowser(url string) { s.openBrowser(url) } + +// Serve runs the wizard until ctx is cancelled. +func (s *Server) Serve(ctx context.Context, listener net.Listener) error { + mux := http.NewServeMux() + mux.HandleFunc("GET /", s.handleIndex) + mux.HandleFunc("GET /api/status", s.handleStatus) + mux.HandleFunc("POST /api/config", s.handleSaveConfig) + mux.HandleFunc("POST /api/test", s.handleTest) + mux.HandleFunc("POST /api/start", s.handleStart) + mux.HandleFunc("POST /api/stop", s.handleStop) + mux.HandleFunc("GET /api/logs", s.handleLogs) + server := &http.Server{ + Handler: mux, + ReadHeaderTimeout: 10 * time.Second, + } + go func() { + <-ctx.Done() + shutdownCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 3*time.Second) + defer cancel() + _ = server.Shutdown(shutdownCtx) + }() + return server.Serve(listener) +} + +func (s *Server) handleIndex(w http.ResponseWriter, r *http.Request) { + if r.URL.Path != "/" { + http.NotFound(w, r) + return + } + html, err := templateFS.ReadFile("template.html") + if err != nil { + http.Error(w, "template unavailable", http.StatusInternalServerError) + return + } + w.Header().Set("Content-Type", "text/html; charset=utf-8") + _, _ = w.Write(html) +} + +func writeJSON(w http.ResponseWriter, status int, v any) { + w.Header().Set("Content-Type", "application/json") + w.WriteHeader(status) + _ = json.NewEncoder(w).Encode(v) +} + +// statusView is what the wizard needs to paint the running/stopped state. +type statusView struct { + ConfigPresent bool `json:"config_present"` + ConfigPath string `json:"config_path"` + LogPath string `json:"log_path"` + Running bool `json:"running"` + Pid int `json:"pid"` + WorkerName string `json:"worker_name,omitempty"` + Coordinator string `json:"coordinator,omitempty"` + WorkDir string `json:"work_dir,omitempty"` + TokenSet bool `json:"token_set"` +} + +func (s *Server) handleStatus(w http.ResponseWriter, r *http.Request) { + view := statusView{ConfigPath: s.cfgPath, LogPath: s.logPath, Running: s.sup.Alive(), Pid: s.sup.Pid()} + if raw, err := os.ReadFile(s.cfgPath); err == nil { + var file agent.ConfigFile + if json.Unmarshal(raw, &file) == nil { + view.ConfigPresent = true + view.WorkerName = file.WorkerName + view.Coordinator = file.CoordinatorURL + view.WorkDir = file.WorkDir + view.TokenSet = file.Token != "" || file.WorkerKey != "" + } + } + writeJSON(w, http.StatusOK, view) +} + +type saveConfigRequest struct { + CoordinatorURL string `json:"coordinator_url"` + Token string `json:"token"` + WorkerKey string `json:"worker_key"` + UserserviceURL string `json:"userservice_url"` + WorkDir string `json:"work_dir"` + WorkerName string `json:"worker_name"` + CPUCount int `json:"cpu_count"` + MemoryMB int `json:"memory_mb"` + TaskRunner []string `json:"task_runner"` +} + +func (s *Server) handleSaveConfig(w http.ResponseWriter, r *http.Request) { + var req saveConfigRequest + if err := json.NewDecoder(io.LimitReader(r.Body, 1<<20)).Decode(&req); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON body"}) + return + } + file := agent.ConfigFile{ + CoordinatorURL: strings.TrimSpace(req.CoordinatorURL), + Token: req.Token, + WorkerKey: req.WorkerKey, + UserserviceURL: strings.TrimSpace(req.UserserviceURL), + WorkDir: strings.TrimSpace(req.WorkDir), + WorkerName: strings.TrimSpace(req.WorkerName), + CPUCount: req.CPUCount, + MemoryMB: req.MemoryMB, + TaskRunner: req.TaskRunner, + } + if file.CoordinatorURL == "" { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "coordinator_url is required"}) + return + } + if !strings.HasPrefix(file.CoordinatorURL, "http://") && !strings.HasPrefix(file.CoordinatorURL, "https://") { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "coordinator_url must be an absolute HTTP(S) URL"}) + return + } + if file.WorkDir == "" { + file.WorkDir = "./scimesh-agent-data" + } + if file.WorkerName == "" { + if host, err := os.Hostname(); err == nil { + file.WorkerName = host + } else { + file.WorkerName = "worker" + } + } + if file.CPUCount < 1 { + file.CPUCount = 1 + } + if err := agent.SaveConfigFile(s.cfgPath, file); err != nil { + s.log.Error("save wizard config", "err", err) + writeJSON(w, http.StatusInternalServerError, map[string]string{"error": "could not write the config file"}) + return + } + writeJSON(w, http.StatusOK, map[string]bool{"saved": true}) +} + +func (s *Server) handleTest(w http.ResponseWriter, r *http.Request) { + var req saveConfigRequest + if err := json.NewDecoder(io.LimitReader(r.Body, 1<<20)).Decode(&req); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "invalid JSON body"}) + return + } + url := strings.TrimSpace(req.CoordinatorURL) + if url == "" { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "coordinator_url is required"}) + return + } + report := agent.RunCheck(r.Context(), url) + writeJSON(w, http.StatusOK, report) +} + +func (s *Server) handleStart(w http.ResponseWriter, r *http.Request) { + if _, err := os.Stat(s.cfgPath); err != nil { + writeJSON(w, http.StatusBadRequest, map[string]string{"error": "no configuration saved yet"}) + return + } + pid, err := s.sup.Start(s.cfgPath, s.logPath) + if err != nil { + writeJSON(w, http.StatusConflict, map[string]string{"error": err.Error()}) + return + } + writeJSON(w, http.StatusOK, map[string]int{"pid": pid}) +} + +func (s *Server) handleStop(w http.ResponseWriter, r *http.Request) { + if err := s.sup.Stop(); err != nil { + writeJSON(w, http.StatusConflict, map[string]string{"error": err.Error()}) + return + } + writeJSON(w, http.StatusOK, map[string]bool{"stopped": true}) +} + +func (s *Server) handleLogs(w http.ResponseWriter, r *http.Request) { + raw, err := os.ReadFile(s.logPath) + if err != nil { + writeJSON(w, http.StatusOK, map[string]string{"log": ""}) + return + } + lines := strings.Split(strings.TrimRight(string(raw), "\n"), "\n") + tail := 200 + if n, err := strconv.Atoi(r.URL.Query().Get("tail")); err == nil && n > 0 && n < 5000 { + tail = n + } + if len(lines) > tail { + lines = lines[len(lines)-tail:] + } + writeJSON(w, http.StatusOK, map[string]string{"log": strings.Join(lines, "\n")}) +} + +// ErrCanceled mirrors context.Canceled for callers that treat a cancelled +// wizard as a clean exit. +var ErrCanceled = errors.New("setup wizard cancelled") diff --git a/coordinator/internal/agent/setupui/server_test.go b/coordinator/internal/agent/setupui/server_test.go new file mode 100644 index 0000000..69cf72a --- /dev/null +++ b/coordinator/internal/agent/setupui/server_test.go @@ -0,0 +1,272 @@ +package setupui + +import ( + "context" + "encoding/json" + "io" + "log/slog" + "net/http" + "net/http/httptest" + "os" + "path/filepath" + "strings" + "sync" + "testing" + "time" + + "github.com/emil28092005/SciMesh/coordinator/internal/agent" +) + +func testLogger() *slog.Logger { + return slog.New(slog.NewTextHandler(io.Discard, nil)) +} + +func newTestServer(t *testing.T, sup Supervisor) (*Server, string) { + t.Helper() + dir := t.TempDir() + server := New(testLogger(), Options{ + ConfigPath: filepath.Join(dir, "config.json"), + Dir: dir, + Supervisor: sup, + OpenBrowser: func(string) {}, + }) + listener, err := server.Listen() + if err != nil { + t.Fatal(err) + } + t.Cleanup(func() { _ = listener.Close() }) + ctx, cancel := context.WithCancel(context.Background()) + t.Cleanup(cancel) + go func() { _ = server.Serve(ctx, listener) }() + return server, "http://" + listener.Addr().String() +} + +type fakeSup struct { + mu sync.Mutex + started bool + stopped bool + pid int + alive bool +} + +func (f *fakeSup) Start(configPath, logPath string) (int, error) { + f.mu.Lock() + defer f.mu.Unlock() + f.started = true + f.alive = true + f.pid = 4242 + return f.pid, nil +} + +func (f *fakeSup) Stop() error { + f.mu.Lock() + defer f.mu.Unlock() + f.stopped = true + f.alive = false + return nil +} + +func (f *fakeSup) Pid() int { f.mu.Lock(); defer f.mu.Unlock(); return f.pid } +func (f *fakeSup) Alive() bool { + f.mu.Lock() + defer f.mu.Unlock() + return f.alive +} + +func postJSON(t *testing.T, base, path string, body any) (*httptest.ResponseRecorder, map[string]any) { + t.Helper() + req, err := http.NewRequest(http.MethodPost, base+path, strings.NewReader(mustJSON(t, body))) + if err != nil { + t.Fatal(err) + } + req.Header.Set("Content-Type", "application/json") + client := http.Client{} + resp, err := client.Do(req) + if err != nil { + t.Fatal(err) + } + defer func() { _ = resp.Body.Close() }() + rec := httptest.NewRecorder() + rec.Code = resp.StatusCode + data := map[string]any{} + _ = json.NewDecoder(resp.Body).Decode(&data) + return rec, data +} + +func mustJSON(t *testing.T, v any) string { + t.Helper() + raw, err := json.Marshal(v) + if err != nil { + t.Fatal(err) + } + return string(raw) +} + +func TestWizardSavesConfigWithStrictPermissions(t *testing.T) { + sup := &fakeSup{} + server, base := newTestServer(t, sup) + + rec, _ := postJSON(t, base, "/api/config", map[string]any{ + "coordinator_url": "http://192.168.1.10:8080", + "token": "sm_live_secret", + "work_dir": "/home/emil/scimesh-worker", + "worker_name": "emil-laptop", + "cpu_count": 8, + "memory_mb": 16384, + }) + if rec.Code != http.StatusOK { + t.Fatalf("save config: got %d, want 200", rec.Code) + } + info, err := os.Stat(server.cfgPath) + if err != nil { + t.Fatal(err) + } + if perm := info.Mode().Perm(); perm != 0o600 { + t.Errorf("config perms = %o, want 600", perm) + } + config, err := agent.LoadConfigFile(server.cfgPath) + if err != nil { + t.Fatal(err) + } + if config.CoordinatorURL != "http://192.168.1.10:8080" || config.Token != "sm_live_secret" || config.WorkDir != "/home/emil/scimesh-worker" || config.WorkerName != "emil-laptop" { + t.Errorf("config = %+v", config) + } + if config.CPUCount != 8 || config.MemoryMB != 16384 { + t.Errorf("resources: cpu=%d mem=%d", config.CPUCount, config.MemoryMB) + } +} + +func TestWizardRejectsInvalidConfig(t *testing.T) { + sup := &fakeSup{} + _, base := newTestServer(t, sup) + + for _, body := range []map[string]any{ + {"coordinator_url": "", "token": "x"}, + {"coordinator_url": "not-a-url", "token": "x"}, + } { + rec, _ := postJSON(t, base, "/api/config", body) + if rec.Code != http.StatusBadRequest { + t.Errorf("body %v: got %d, want 400", body, rec.Code) + } + } +} + +func TestWizardStartStopLifecycle(t *testing.T) { + sup := &fakeSup{} + _, base := newTestServer(t, sup) + + // Starting without a saved config is rejected. + rec, _ := postJSON(t, base, "/api/start", map[string]any{}) + if rec.Code != http.StatusBadRequest { + t.Errorf("start without config: got %d, want 400", rec.Code) + } + + postJSON(t, base, "/api/config", map[string]any{ + "coordinator_url": "http://127.0.0.1:8080", "token": "t", "work_dir": ".", + }) + rec, data := postJSON(t, base, "/api/start", map[string]any{}) + if rec.Code != http.StatusOK || int(data["pid"].(float64)) != 4242 { + t.Errorf("start: got %d %v, want 200 pid 4242", rec.Code, data) + } + if !sup.started { + t.Error("supervisor never started the worker") + } + + // Status reflects the running state. + req, _ := http.NewRequest(http.MethodGet, base+"/api/status", nil) + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer func() { _ = resp.Body.Close() }() + var status map[string]any + _ = json.NewDecoder(resp.Body).Decode(&status) + if status["running"] != true || status["pid"] != float64(4242) { + t.Errorf("status = %v, want running pid 4242", status) + } + + rec, _ = postJSON(t, base, "/api/stop", map[string]any{}) + if rec.Code != http.StatusOK || !sup.stopped { + t.Errorf("stop: got %d stopped=%v, want 200/true", rec.Code, sup.stopped) + } +} + +func TestWizardStatusPrefillsSavedConfig(t *testing.T) { + sup := &fakeSup{} + _, base := newTestServer(t, sup) + postJSON(t, base, "/api/config", map[string]any{ + "coordinator_url": "http://10.0.0.5:8080", "worker_key": "smk_abc", "work_dir": "/w", "worker_name": "n1", + }) + req, _ := http.NewRequest(http.MethodGet, base+"/api/status", nil) + resp, err := http.DefaultClient.Do(req) + if err != nil { + t.Fatal(err) + } + defer func() { _ = resp.Body.Close() }() + var status struct { + ConfigPresent bool `json:"config_present"` + WorkerName string `json:"worker_name"` + Coordinator string `json:"coordinator"` + TokenSet bool `json:"token_set"` + } + _ = json.NewDecoder(resp.Body).Decode(&status) + if !status.ConfigPresent || status.WorkerName != "n1" || status.Coordinator != "http://10.0.0.5:8080" || !status.TokenSet { + t.Errorf("status = %+v", status) + } + // The secret must never appear in the status projection. + if strings.Contains(strings.ToLower(mustJSON(t, status)), "smk_abc") { + t.Error("status leaks the worker key") + } +} + +func TestCheckCoordinatorReachable(t *testing.T) { + stub := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + if r.URL.Path == "/health" { + _, _ = w.Write([]byte(`{"status":"ok"}`)) + return + } + http.NotFound(w, r) + })) + defer stub.Close() + + report := agent.CheckCoordinator(context.Background(), stub.URL, 5*time.Second) + if !report.Coordinator.OK { + t.Errorf("coordinator check = %+v, want ok", report.Coordinator) + } +} + +func TestCheckCoordinatorUnreachable(t *testing.T) { + report := agent.CheckCoordinator(context.Background(), "http://127.0.0.1:1", 2*time.Second) + if report.Coordinator.OK { + t.Error("unreachable coordinator reported ok") + } +} + +func TestConfigFileDefaultsAndEnvOverride(t *testing.T) { + dir := t.TempDir() + path := filepath.Join(dir, "config.json") + file := agent.ConfigFile{ + CoordinatorURL: "http://coord:8080", + Token: "file-token", + WorkDir: "/w", + CPUCount: 4, + } + if err := agent.SaveConfigFile(path, file); err != nil { + t.Fatal(err) + } + t.Setenv("COORDINATOR_URL", "http://env:9090") + t.Setenv("WORKER_AUTH_TOKEN", "") + config, err := agent.LoadConfigFile(path) + if err != nil { + t.Fatal(err) + } + if config.CoordinatorURL != "http://env:9090" { + t.Errorf("env must win: %s", config.CoordinatorURL) + } + if config.Token != "file-token" { + t.Errorf("token = %q, want the file value", config.Token) + } + if config.CPUCount != 4 { + t.Errorf("cpu = %d", config.CPUCount) + } +} diff --git a/coordinator/internal/agent/setupui/template.html b/coordinator/internal/agent/setupui/template.html new file mode 100644 index 0000000..a1d93af --- /dev/null +++ b/coordinator/internal/agent/setupui/template.html @@ -0,0 +1,310 @@ + + + + + +SciMesh Worker · Setup + + + +
+
+
+
SciMesh · Worker setup
+
+

Local wizard served by worker-agent setup · 127.0.0.1

+ + +
+
+
1
Connect
+
+
2
Machine
+
+
3
Check
+
+
4
Run
+
+
+ + +
+
+

Connect to a coordinator

+

The coordinator hands out work and collects results. Ask your cluster admin for its address.

+

For a served instance this is the address printed by coordinator serve.

+
+
+
Cluster token

Serve instances: one token for every worker.

+
Worker key

Shared clusters: a key tied to your account.

+
+
+

The admin can copy it from coordinator token on the server.

+ +
Step 1 of 4
+
+
+ + +
+
+

This machine

+

Where tasks run and how the machine appears in the cluster.

+

Shown in the coordinator’s worker list.

+

Datasets and shard results live here. ~1 GB free space recommended.

+
+
+
Auto

Detect the machine’s CPU count.

+
Custom…

Limit what this machine advertises.

+
+
+ +
+
+
+ + +
+
+

Preflight check

+

Making sure this machine can reach the coordinator and run SciMesh workloads.

+
+
+
+
+ + +
+
+
+

Ready to join the cluster

+

Configuration will be saved and the worker started as a background process.

+
+
+
+
+ + + +
+ + + +