448 lines
11 KiB
Go
448 lines
11 KiB
Go
// Package nodecontrol exposes fixed host operations over a local Unix socket.
|
|
package nodecontrol
|
|
|
|
import (
|
|
"context"
|
|
"crypto/rand"
|
|
"encoding/hex"
|
|
"encoding/json"
|
|
"errors"
|
|
"fmt"
|
|
"io"
|
|
"net"
|
|
"net/http"
|
|
"net/netip"
|
|
"os"
|
|
"path/filepath"
|
|
"regexp"
|
|
"sort"
|
|
"strings"
|
|
"sync"
|
|
"time"
|
|
)
|
|
|
|
const Socket = "/run/felis-node-control/control.sock"
|
|
const MaxLog = 64 << 10
|
|
|
|
var ErrBusy = errors.New("a node operation is already running")
|
|
var ErrNotFound = errors.New("node operation not found")
|
|
var namePattern = regexp.MustCompile(`^[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?$`)
|
|
var targetPattern = regexp.MustCompile(`^[a-zA-Z0-9][a-zA-Z0-9_.@:-]{0,252}$`)
|
|
var idPattern = regexp.MustCompile(`^[a-f0-9]{32}$`)
|
|
|
|
func validNodeName(name string) bool {
|
|
if len(name) > 63 {
|
|
return false
|
|
}
|
|
for _, label := range strings.Split(name, ".") {
|
|
if !namePattern.MatchString(label) {
|
|
return false
|
|
}
|
|
}
|
|
return true
|
|
}
|
|
|
|
type Request struct {
|
|
Action string `json:"action"`
|
|
Name string `json:"name,omitempty"`
|
|
SSHTarget string `json:"sshTarget,omitempty"`
|
|
ExternalIP string `json:"externalIP,omitempty"`
|
|
Peers []string `json:"peers,omitempty"`
|
|
ConfirmMaintenance bool `json:"confirmMaintenance"`
|
|
}
|
|
|
|
func (r Request) Validate() error {
|
|
if !r.ConfirmMaintenance {
|
|
return errors.New("maintenance impact confirmation is required")
|
|
}
|
|
switch r.Action {
|
|
case "join", "approve":
|
|
if !validNodeName(r.Name) || !targetPattern.MatchString(r.SSHTarget) {
|
|
return errors.New("invalid worker name or SSH target")
|
|
}
|
|
case "enable":
|
|
default:
|
|
return errors.New("unsupported node operation")
|
|
}
|
|
if r.Action != "approve" {
|
|
if a, err := netip.ParseAddr(r.ExternalIP); err != nil || a.IsUnspecified() || a.IsLoopback() || a.IsMulticast() || a.Zone() != "" {
|
|
return errors.New("a fixed node IP is required")
|
|
}
|
|
}
|
|
for _, peer := range r.Peers {
|
|
p, err := netip.ParsePrefix(peer)
|
|
if err != nil || p.Bits() != p.Addr().BitLen() || p.Addr().IsLoopback() || p.Addr().IsMulticast() || p.Addr().IsUnspecified() {
|
|
return errors.New("peer addresses must use exact /32 or /128 prefixes")
|
|
}
|
|
}
|
|
if len(r.Peers) > 100 {
|
|
return errors.New("too many peer addresses")
|
|
}
|
|
return nil
|
|
}
|
|
|
|
type Task struct {
|
|
ID string `json:"id"`
|
|
Request Request `json:"request"`
|
|
Actor string `json:"actor"`
|
|
State string `json:"state"`
|
|
Stage string `json:"stage"`
|
|
StartedAt time.Time `json:"startedAt"`
|
|
FinishedAt *time.Time `json:"finishedAt,omitempty"`
|
|
Error string `json:"error,omitempty"`
|
|
Log string `json:"log,omitempty"`
|
|
}
|
|
|
|
type Executor func(context.Context, Request, func(string) error, io.Writer) error
|
|
|
|
type Manager struct {
|
|
dir string
|
|
execute Executor
|
|
mu sync.Mutex
|
|
tasks map[string]Task
|
|
active string
|
|
ctx context.Context
|
|
wg sync.WaitGroup
|
|
}
|
|
|
|
func Open(ctx context.Context, dir string, execute Executor) (*Manager, error) {
|
|
if err := os.MkdirAll(dir, 0700); err != nil {
|
|
return nil, err
|
|
}
|
|
m := &Manager{dir: dir, execute: execute, tasks: map[string]Task{}, ctx: ctx}
|
|
paths, err := filepath.Glob(filepath.Join(dir, "*.json"))
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
for _, path := range paths {
|
|
raw, err := os.ReadFile(path)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
var t Task
|
|
if err = json.Unmarshal(raw, &t); err != nil {
|
|
return nil, fmt.Errorf("read node task: %w", err)
|
|
}
|
|
if !idPattern.MatchString(t.ID) || filepath.Base(path) != t.ID+".json" {
|
|
return nil, errors.New("invalid stored task ID")
|
|
}
|
|
if t.State == "running" {
|
|
t.State, t.Error = "failed", "Host execution service restarted. Verify the host state before retrying."
|
|
now := time.Now().UTC()
|
|
t.FinishedAt = &now
|
|
if err = m.persist(t); err != nil {
|
|
return nil, err
|
|
}
|
|
}
|
|
m.tasks[t.ID] = t
|
|
}
|
|
return m, nil
|
|
}
|
|
func (m *Manager) persist(t Task) error {
|
|
t.Log = ""
|
|
raw, err := json.Marshal(t)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
path := filepath.Join(m.dir, t.ID+".json")
|
|
f, err := os.OpenFile(path+".tmp", os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0600)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if _, err = f.Write(raw); err == nil {
|
|
err = f.Sync()
|
|
}
|
|
closeErr := f.Close()
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if closeErr != nil {
|
|
return closeErr
|
|
}
|
|
if err = os.Rename(path+".tmp", path); err != nil {
|
|
return err
|
|
}
|
|
d, err := os.Open(m.dir)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer d.Close()
|
|
return d.Sync()
|
|
}
|
|
func (m *Manager) List() []Task {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
out := make([]Task, 0, len(m.tasks))
|
|
for _, t := range m.tasks {
|
|
out = append(out, t)
|
|
}
|
|
sort.Slice(out, func(i, j int) bool { return out[i].StartedAt.After(out[j].StartedAt) })
|
|
if len(out) > 100 {
|
|
out = out[:100]
|
|
}
|
|
return out
|
|
}
|
|
func (m *Manager) Get(id string) (Task, error) {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
t, ok := m.tasks[id]
|
|
if !ok {
|
|
return Task{}, ErrNotFound
|
|
}
|
|
f, err := os.Open(filepath.Join(m.dir, id+".log"))
|
|
if errors.Is(err, os.ErrNotExist) {
|
|
return t, nil
|
|
}
|
|
if err != nil {
|
|
return t, err
|
|
}
|
|
defer f.Close()
|
|
info, err := f.Stat()
|
|
if err != nil {
|
|
return t, err
|
|
}
|
|
if info.Size() > MaxLog {
|
|
if _, err = f.Seek(-MaxLog, io.SeekEnd); err != nil {
|
|
return t, err
|
|
}
|
|
}
|
|
raw, err := io.ReadAll(f)
|
|
t.Log = string(raw)
|
|
return t, err
|
|
}
|
|
func (m *Manager) Start(r Request, actor string) (Task, error) {
|
|
if err := r.Validate(); err != nil {
|
|
return Task{}, err
|
|
}
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
if m.active != "" {
|
|
return Task{}, ErrBusy
|
|
}
|
|
if err := m.ctx.Err(); err != nil {
|
|
return Task{}, err
|
|
}
|
|
var id [16]byte
|
|
if _, err := rand.Read(id[:]); err != nil {
|
|
return Task{}, err
|
|
}
|
|
t := Task{ID: hex.EncodeToString(id[:]), Request: r, Actor: actor, State: "running", Stage: "preflight", StartedAt: time.Now().UTC()}
|
|
if err := m.persist(t); err != nil {
|
|
return Task{}, err
|
|
}
|
|
m.tasks[t.ID] = t
|
|
m.active = t.ID
|
|
m.wg.Add(1)
|
|
go m.run(t)
|
|
return t, nil
|
|
}
|
|
func (m *Manager) run(t Task) {
|
|
defer m.wg.Done()
|
|
ctx, cancel := context.WithTimeout(m.ctx, 45*time.Minute)
|
|
defer cancel()
|
|
log, err := os.OpenFile(filepath.Join(m.dir, t.ID+".log"), os.O_CREATE|os.O_RDWR|os.O_APPEND, 0600)
|
|
if err == nil {
|
|
stage := func(s string) error {
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
t.Stage = s
|
|
if err := m.persist(t); err != nil {
|
|
return err
|
|
}
|
|
m.tasks[t.ID] = t
|
|
_, err := fmt.Fprintln(log, "[felis]", s)
|
|
return err
|
|
}
|
|
err = m.execute(ctx, t.Request, stage, &boundedLog{file: log})
|
|
if ctx.Err() != nil {
|
|
err = ctx.Err()
|
|
}
|
|
if closeErr := log.Close(); err == nil {
|
|
err = closeErr
|
|
}
|
|
}
|
|
m.mu.Lock()
|
|
defer m.mu.Unlock()
|
|
now := time.Now().UTC()
|
|
t.FinishedAt = &now
|
|
t.State = "succeeded"
|
|
if err != nil {
|
|
t.State = "failed"
|
|
t.Error = err.Error()
|
|
if errors.Is(err, context.DeadlineExceeded) {
|
|
t.Error = "Host operation exceeded the 45-minute execution limit. Verify the host state before retrying."
|
|
}
|
|
if errors.Is(err, context.Canceled) {
|
|
t.Error = "Host operation was interrupted. Verify the host state before retrying."
|
|
}
|
|
}
|
|
if persistErr := m.persist(t); persistErr != nil {
|
|
t.State = "failed"
|
|
t.Error = "Task result persistence failed: " + persistErr.Error()
|
|
}
|
|
m.tasks[t.ID] = t
|
|
m.active = ""
|
|
for len(m.tasks) > 100 {
|
|
oldest := ""
|
|
for id, candidate := range m.tasks {
|
|
if id == t.ID {
|
|
continue
|
|
}
|
|
if oldest == "" || candidate.StartedAt.Before(m.tasks[oldest].StartedAt) {
|
|
oldest = id
|
|
}
|
|
}
|
|
if oldest != "" {
|
|
if err := os.Remove(filepath.Join(m.dir, oldest+".json")); err == nil {
|
|
os.Remove(filepath.Join(m.dir, oldest+".log"))
|
|
delete(m.tasks, oldest)
|
|
} else {
|
|
break
|
|
}
|
|
}
|
|
}
|
|
}
|
|
func (m *Manager) Wait() { m.wg.Wait() }
|
|
|
|
// Handler is reachable only through the root-owned socket mounted into felis-api.
|
|
func (m *Manager) Handler() http.Handler {
|
|
mux := http.NewServeMux()
|
|
mux.HandleFunc("GET /tasks", func(w http.ResponseWriter, r *http.Request) { json.NewEncoder(w).Encode(m.List()) })
|
|
mux.HandleFunc("GET /tasks/{id}", func(w http.ResponseWriter, r *http.Request) {
|
|
t, err := m.Get(r.PathValue("id"))
|
|
if err != nil {
|
|
http.Error(w, err.Error(), 404)
|
|
return
|
|
}
|
|
json.NewEncoder(w).Encode(t)
|
|
})
|
|
mux.HandleFunc("POST /tasks", func(w http.ResponseWriter, r *http.Request) {
|
|
var body struct {
|
|
Request Request `json:"request"`
|
|
Actor string `json:"actor"`
|
|
}
|
|
dec := json.NewDecoder(http.MaxBytesReader(w, r.Body, 8192))
|
|
dec.DisallowUnknownFields()
|
|
if err := dec.Decode(&body); err != nil {
|
|
http.Error(w, "invalid task request", 400)
|
|
return
|
|
}
|
|
if err := dec.Decode(&struct{}{}); err != io.EOF {
|
|
http.Error(w, "invalid trailing task data", 400)
|
|
return
|
|
}
|
|
if err := body.Request.Validate(); err != nil {
|
|
http.Error(w, err.Error(), 400)
|
|
return
|
|
}
|
|
t, err := m.Start(body.Request, body.Actor)
|
|
if err != nil {
|
|
code := 500
|
|
if errors.Is(err, ErrBusy) {
|
|
code = 409
|
|
}
|
|
http.Error(w, err.Error(), code)
|
|
return
|
|
}
|
|
w.WriteHeader(http.StatusAccepted)
|
|
json.NewEncoder(w).Encode(t)
|
|
})
|
|
return mux
|
|
}
|
|
|
|
type Client struct{ http *http.Client }
|
|
|
|
func NewClient(socket string) *Client {
|
|
return &Client{http: &http.Client{Timeout: 10 * time.Second, Transport: &http.Transport{DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) {
|
|
return (&net.Dialer{}).DialContext(ctx, "unix", socket)
|
|
}}}}
|
|
}
|
|
func (c *Client) call(ctx context.Context, method, path string, body any, out any) error {
|
|
var reader io.Reader
|
|
if body != nil {
|
|
raw, err := json.Marshal(body)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
reader = strings.NewReader(string(raw))
|
|
}
|
|
req, err := http.NewRequestWithContext(ctx, method, "http://node-control"+path, reader)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
res, err := c.http.Do(req)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
defer res.Body.Close()
|
|
if res.StatusCode == 409 {
|
|
return ErrBusy
|
|
}
|
|
if res.StatusCode == 404 {
|
|
return ErrNotFound
|
|
}
|
|
if res.StatusCode >= 400 {
|
|
return fmt.Errorf("host service returned HTTP %d", res.StatusCode)
|
|
}
|
|
return json.NewDecoder(io.LimitReader(res.Body, 1<<20)).Decode(out)
|
|
}
|
|
func (c *Client) List(ctx context.Context) ([]Task, error) {
|
|
var out []Task
|
|
err := c.call(ctx, "GET", "/tasks", nil, &out)
|
|
return out, err
|
|
}
|
|
func (c *Client) Get(ctx context.Context, id string) (Task, error) {
|
|
if !idPattern.MatchString(id) {
|
|
return Task{}, ErrNotFound
|
|
}
|
|
var out Task
|
|
err := c.call(ctx, "GET", "/tasks/"+id, nil, &out)
|
|
return out, err
|
|
}
|
|
func (c *Client) Start(ctx context.Context, r Request, actor string) (Task, error) {
|
|
var out Task
|
|
err := c.call(ctx, "POST", "/tasks", map[string]any{"request": r, "actor": actor}, &out)
|
|
return out, err
|
|
}
|
|
|
|
// Bound disk use while retaining the most recent output of long builds.
|
|
type boundedLog struct {
|
|
file *os.File
|
|
mu sync.Mutex
|
|
}
|
|
|
|
func (w *boundedLog) Write(p []byte) (int, error) {
|
|
w.mu.Lock()
|
|
defer w.mu.Unlock()
|
|
const limit = 4 << 20
|
|
info, err := w.file.Stat()
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
if info.Size()+int64(len(p)) > limit {
|
|
n := int64(MaxLog)
|
|
if info.Size() < n {
|
|
n = info.Size()
|
|
}
|
|
tail := make([]byte, n)
|
|
if _, err = w.file.ReadAt(tail, info.Size()-n); err != nil {
|
|
return 0, err
|
|
}
|
|
if err = w.file.Truncate(0); err != nil {
|
|
return 0, err
|
|
}
|
|
if _, err = w.file.Write(tail); err != nil {
|
|
return 0, err
|
|
}
|
|
}
|
|
original := len(p)
|
|
if len(p) > limit {
|
|
p = p[len(p)-limit:]
|
|
}
|
|
_, err = w.file.Write(p)
|
|
if err != nil {
|
|
return 0, err
|
|
}
|
|
return original, nil
|
|
}
|