nmon/internal/vmsup/supervisor.go

package vmsup

import (
	"context"
	"errors"
	"fmt"
	"log/slog"
	"net/http"
	"net/url"
	"os"
	"strings"
	"sync"
	"time"

	"github.com/v10networks/nmon/internal/clock"
	"github.com/v10networks/nmon/internal/config"
)

// Store names.
const (
	StoreHot  = "hot"
	StoreLong = "long"
)

// Options configures a Supervisor.
type Options struct {
	DataDir string
	// HotURL and LongURL, when non-empty, name an external instance: no child
	// is run for that store and nothing but /health is probed.
	HotURL, LongURL             string
	HotRetention, LongRetention time.Duration // defaults: 30 days, 2 years
	Clock                       clock.Clock   // default clock.Real()
	Logger                      *slog.Logger  // default slog.Default()
	// OnEvent receives store_failed, store_restarted and store_recovered. It is
	// called from the supervisor's goroutines and must not block.
	OnEvent func(Event)
	// ExtraArgs adds command-line flags per store ("hot", "long"). They come
	// last, so they override the defaults.
	ExtraArgs map[string][]string

	// Test hooks.
	backoff      time.Duration // default backoffBase
	healthEvery  time.Duration // default healthEvery
	startTimeout time.Duration // default startTimeout
	stopTimeout  time.Duration // default stopTimeout
}

// ChildStatus is a snapshot of one store.
type ChildStatus struct {
	Store     string
	URL       string
	External  bool // no child: an operator-run instance
	Healthy   bool
	Failed    bool // store_failed raised; restarts stopped until Restart
	PID       int  // 0 when no child is running
	Restarts  int  // restarts since Start
	StartedAt time.Time
	LastExit  string
}

// Supervisor runs and watches the hot and long stores.
type Supervisor struct {
	opts   Options
	log    *slog.Logger
	clk    clock.Clock
	stores []*store // hot, long
	hc     *http.Client

	mu      sync.Mutex
	started bool
	stopped bool
	cancel  context.CancelFunc
	killNow chan struct{}
	queues  []*Queue
	spec    binarySpec
}

// withDefaults fills the zero-valued options.
func (o Options) withDefaults() Options {
	if o.Clock == nil {
		o.Clock = clock.Real()
	}
	if o.Logger == nil {
		o.Logger = slog.Default()
	}
	if o.backoff <= 0 {
		o.backoff = backoffBase
	}
	if o.healthEvery <= 0 {
		o.healthEvery = healthEvery
	}
	if o.startTimeout <= 0 {
		o.startTimeout = startTimeout
	}
	if o.stopTimeout <= 0 {
		o.stopTimeout = stopTimeout
	}
	if o.HotRetention <= 0 {
		o.HotRetention = 30 * 24 * time.Hour
	}
	if o.LongRetention <= 0 {
		o.LongRetention = 2 * 365 * 24 * time.Hour
	}
	return o
}

// New validates opts and returns a Supervisor. Nothing runs until Start.
func New(opts Options) (*Supervisor, error) {
	opts = opts.withDefaults()
	s := &Supervisor{
		opts:    opts,
		log:     opts.Logger.With("component", "vmsup"),
		clk:     opts.Clock,
		hc:      &http.Client{Timeout: 2 * time.Second},
		killNow: make(chan struct{}),
	}
	for _, d := range []struct {
		name, ext, dir string
		ret            time.Duration
	}{
		{StoreHot, opts.HotURL, config.HotDir(opts.DataDir), opts.HotRetention},
		{StoreLong, opts.LongURL, config.LongDir(opts.DataDir), opts.LongRetention},
	} {
		st := &store{
			name: d.name, dir: d.dir, retention: d.ret, extra: opts.ExtraArgs[d.name],
			resume: make(chan struct{}, 1), done: make(chan struct{}),
		}
		if err := st.configureExternal(d.ext, opts.DataDir); err != nil {
			return nil, err
		}
		s.stores = append(s.stores, st)
	}
	return s, nil
}

// configureExternal marks the store external when ext names an instance; a
// managed store needs a data directory instead.
func (st *store) configureExternal(ext, dataDir string) error {
	if ext == "" {
		if dataDir == "" {
			return errors.New("vmsup: DataDir is required when a store is not external")
		}
		return nil
	}
	u, err := url.Parse(ext)
	if err != nil || (u.Scheme != "http" && u.Scheme != "https") || u.Host == "" {
		return fmt.Errorf("vmsup: %s store URL %q is not an http(s) URL", st.name, ext)
	}
	st.external = true
	st.url = strings.TrimRight(ext, "/")
	return nil
}

func (s *Supervisor) store(name string) *store {
	for _, st := range s.stores {
		if st.name == name {
			return st
		}
	}
	return nil
}

// Start extracts and verifies the store binary, starts the children, and waits
// until both stores answer /health (30 s). ctx bounds only that wait; the
// children live until Stop. If Start fails nothing is left running.
func (s *Supervisor) Start(ctx context.Context) error {
	s.mu.Lock()
	if s.started || s.stopped {
		s.mu.Unlock()
		return errors.New("vmsup: supervisor already started")
	}
	s.started = true
	s.mu.Unlock()

	if err := s.prepareManaged(); err != nil {
		s.fail()
		return err
	}

	runCtx, cancel := context.WithCancel(context.Background())
	s.mu.Lock()
	s.cancel = cancel
	s.mu.Unlock()
	for _, st := range s.stores {
		if st.external {
			go s.watchExternal(runCtx, st)
		} else {
			go s.supervise(runCtx, st)
		}
	}

	if err := s.waitHealthy(ctx); err != nil {
		_ = s.Stop(context.Background())
		return err
	}
	return nil
}

// prepareManaged verifies the store binary and creates the data directories of
// the stores nmon runs itself. It does nothing when every store is external.
func (s *Supervisor) prepareManaged() error {
	managed := false
	for _, st := range s.stores {
		managed = managed || !st.external
	}
	if !managed {
		return nil
	}
	spec, err := prepareBinary(s.opts.DataDir)
	if err != nil {
		return err
	}
	s.spec = spec
	for _, st := range s.stores {
		if st.external {
			continue
		}
		if err := os.MkdirAll(st.dir, 0o755); err != nil {
			return fmt.Errorf("vmsup: create %s: %w", st.dir, err)
		}
	}
	return nil
}

// waitHealthy blocks until both stores answer /health, one fails, the start
// timeout passes, or ctx ends.
func (s *Supervisor) waitHealthy(ctx context.Context) error {
	deadline := s.clk.After(s.opts.startTimeout)
	for _, st := range s.stores {
		for !st.isHealthy() {
			if st.isFailed() {
				return fmt.Errorf("vmsup: %s store failed to start: %s", st.name, st.lastExitText())
			}
			select {
			case <-deadline:
				return fmt.Errorf("vmsup: %s store not healthy after %s", st.name, s.opts.startTimeout)
			case <-ctx.Done():
				return ctx.Err()
			case <-s.clk.After(50 * time.Millisecond):
			}
		}
	}
	return nil
}

// fail resets state after a Start that never launched anything.
func (s *Supervisor) fail() {
	s.mu.Lock()
	s.started = false
	s.mu.Unlock()
}

// HotURL is the base URL of the hot store, for example "http://127.0.0.1:41234".
func (s *Supervisor) HotURL() string { return s.URL(StoreHot) }

// LongURL is the base URL of the long store.
func (s *Supervisor) LongURL() string { return s.URL(StoreLong) }

// URL returns the base URL of the named store, or "" if unknown or not yet chosen.
func (s *Supervisor) URL(name string) string {
	if st := s.store(name); st != nil {
		st.mu.Lock()
		defer st.mu.Unlock()
		return st.url
	}
	return ""
}

// Healthy reports whether the named store answered its last health probe.
func (s *Supervisor) Healthy(name string) bool {
	if st := s.store(name); st != nil {
		return st.isHealthy()
	}
	return false
}

// Status returns a snapshot of the hot and long stores, in that order.
func (s *Supervisor) Status() []ChildStatus {
	out := make([]ChildStatus, 0, len(s.stores))
	for _, st := range s.stores {
		st.mu.Lock()
		out = append(out, ChildStatus{
			Store: st.name, URL: st.url, External: st.external, Healthy: st.healthy,
			Failed: st.failed, PID: st.pid, Restarts: st.restarts,
			StartedAt: st.startedAt, LastExit: st.lastExit,
		})
		st.mu.Unlock()
	}
	return out
}

// Restart errors.
var (
	// ErrUnknownStore means the name is neither hot nor long.
	ErrUnknownStore = errors.New("vmsup: unknown store")
	// ErrExternalStore means the store is an external instance nmon does not manage.
	ErrExternalStore = errors.New("vmsup: store is external and not managed by nmon")
	// ErrNotFailed means the store has not given up, so there is nothing to resume.
	ErrNotFailed = errors.New("vmsup: store has not failed")
)

// Restart resumes a store after store_failed: the restart history is cleared
// and the child is started again. It is an error for a store that has not failed.
func (s *Supervisor) Restart(name string) error {
	st := s.store(name)
	if st == nil {
		return fmt.Errorf("%w: %q", ErrUnknownStore, name)
	}
	if st.external {
		return fmt.Errorf("%w: store %q", ErrExternalStore, name)
	}
	st.mu.Lock()
	defer st.mu.Unlock()
	if !st.failed {
		return fmt.Errorf("%w: store %q", ErrNotFailed, name)
	}
	st.failed = false
	st.resume <- struct{}{} // buffered 1; only one Restart can pass the check above
	return nil
}

// Stop flushes registered write queues, sends SIGTERM to the children and waits
// for them. A child still running after 30 s, or when ctx ends, is killed.
func (s *Supervisor) Stop(ctx context.Context) error {
	s.mu.Lock()
	if !s.started || s.stopped {
		s.mu.Unlock()
		return nil
	}
	s.stopped = true
	queues := append([]*Queue(nil), s.queues...)
	cancel := s.cancel
	s.mu.Unlock()

	for _, q := range queues {
		if err := q.Flush(ctx); err != nil {
			s.log.Warn("flush on stop failed", "queue", q.name, "err", err, "depth", q.Depth())
		}
	}

	stopped := make(chan struct{})
	defer close(stopped)
	go func() {
		select {
		case <-ctx.Done():
			close(s.killNow)
		case <-stopped:
		}
	}()
	if cancel != nil {
		cancel()
	}
	for _, st := range s.stores {
		if st.external {
			continue
		}
		<-st.done
	}
	for _, st := range s.stores {
		st.setDown("")
	}
	return nil
}

func (s *Supervisor) register(q *Queue) {
	s.mu.Lock()
	s.queues = append(s.queues, q)
	s.mu.Unlock()
}