felhom-controller/controller/cmd/controller/upwatch_wiring.go

package main

import (
	"context"
	"log"
	"sync"
	"time"

	"gitea.dooplex.hu/admin/felhom-controller/internal/agentapi"
	"gitea.dooplex.hu/admin/felhom-controller/internal/backup"
	"gitea.dooplex.hu/admin/felhom-controller/internal/mediawire"
	"gitea.dooplex.hu/admin/felhom-controller/internal/metrics"
	"gitea.dooplex.hu/admin/felhom-controller/internal/notify"
	"gitea.dooplex.hu/admin/felhom-controller/internal/settings"
	"gitea.dooplex.hu/admin/felhom-controller/internal/stacks"
	"gitea.dooplex.hu/admin/felhom-controller/internal/upwatch"
)

// qbLimitShare is qBittorrent's default share of the measured upload line — CC's pick, open for the operator (decide
// row in felhom.eu STATUS): 50 %.
const qbLimitShare = 0.5

// baselineMinMinutes: an app the box cannot ask is judged only once its history holds ~20 hours of minutes.
const baselineMinMinutes = 1200

// newUpwatch builds the upload and CPU watch (internal/upwatch; felhom.eu `08` §9). It does not start it.
func newUpwatch(sett *settings.Settings, stackMgr *stacks.Manager, backupMgr *backup.Manager, metricsStore *metrics.MetricsStore,
	notifier *notify.Notifier, line agentLine, logger *log.Logger) *upwatch.Service {
	box := mediawire.DockerBox{}
	qb := &upwatch.QBittorrent{Box: box}
	tr := &upwatch.Transmission{Box: box}
	svc := &upwatch.Service{Store: sett, QB: qb, QBRunning: func() bool { return box.Running("qbittorrent") }, QBShare: qbLimitShare,
		Measured: line.Measured, Log: logger,
		// R-1250: a box that seeds all day is line-tested by holding its torrent app low for one test on request.
		Torrents: []upwatch.Holdable{qb, tr}, TorrentRunning: box.Running, LineWaiting: line.Waiting, LineRun: line.Run}
	w := &upwatch.Watcher{
		Source: &upwatch.DockerAPI{},
		Explainers: map[string]upwatch.Explainer{
			"qbittorrent":  qb,
			"transmission": tr,
			"plex":         &upwatch.Plex{Token: func() string { return mediawire.PlexToken(box) }, URL: mediawire.PlexURL},
		},
		Rules: upwatch.DefaultRules(),
		Line:  svc.Line,
		Busy: func(stack string) bool {
			if backupMgr != nil && backupMgr.IsRunning() {
				return true
			}
			if st, ok := stackMgr.GetStack(stack); ok && st != nil && (st.Updating || st.Deploying) {
				return true
			}
			return false
		},
		Emit: notifier.NotifyUnexplained,
		Log:  logger,
	}
	if metricsStore != nil {
		w.Baseline = dailyBaseline(metricsStore)
	}
	svc.Watcher = w
	return svc
}

// agentLine is the host's line test as the watch uses it; every field nil without a local API.
type agentLine struct {
	Measured func(context.Context) (float64, time.Time, bool)
	Waiting  func(context.Context) bool                           // waiting_for_quiet (agent ≥ 0.165.0, R-1250)
	Run      func(context.Context) (upwatch.LineRunResult, error) // POST /host/line-test/run
}

// agentLineTest reads the host's weekly line test through the agent's local API (agent ≥ 0.164.0; an older agent
// answers 404 → not measured, and never waiting). Empty without a local API.
func agentLineTest(endpoint, token, fingerprint string, logger *log.Logger) agentLine {
	if endpoint == "" || token == "" {
		return agentLine{}
	}
	ac, err := agentapi.New(endpoint, token, fingerprint)
	if err != nil {
		logger.Printf("[WARN] [upwatch] agent client init failed (%v) — no line test; qBittorrent gets no limit", err)
		return agentLine{}
	}
	read := func(ctx context.Context) (agentapi.LineTestState, error) {
		cctx, cancel := context.WithTimeout(ctx, 15*time.Second)
		defer cancel()
		return ac.LineTest(cctx)
	}
	return agentLine{
		Measured: func(ctx context.Context) (float64, time.Time, bool) {
			st, err := read(ctx)
			if err != nil {
				logger.Printf("[DEBUG] [upwatch] line test read: %v", err)
				return 0, time.Time{}, false
			}
			return st.Mbps, st.MeasuredAtTime(), st.Present
		},
		Waiting: func(ctx context.Context) bool {
			st, err := read(ctx)
			return err == nil && st.WaitingForQuiet
		},
		Run: func(ctx context.Context) (upwatch.LineRunResult, error) {
			r, err := ac.LineTestRun(ctx)
			return upwatch.LineRunResult{OK: r.OK, Mbps: r.Mbps, Refused: r.Refused, Error: r.Error}, err
		},
	}
}

// dailyBaseline caches each app's p95 for a day (a 7-day query is a few hundred thousand rows).
func dailyBaseline(ms *metrics.MetricsStore) func(string, []string) (float64, bool) {
	type entry struct {
		at  time.Time
		p95 float64
		ok  bool
	}
	var mu sync.Mutex
	cache := map[string]entry{}
	return func(stack string, containers []string) (float64, bool) {
		mu.Lock()
		defer mu.Unlock()
		if e, ok := cache[stack]; ok && time.Since(e.at) < 24*time.Hour {
			return e.p95, e.ok
		}
		p95, minutes, err := ms.StackTxP95Mbps(containers, time.Now().Add(-7*24*time.Hour))
		e := entry{at: time.Now(), p95: p95, ok: err == nil && minutes >= baselineMinMinutes}
		cache[stack] = e
		return e.p95, e.ok
	}
}

// startUpwatch runs the minute watcher and the 15-minute qBittorrent limit rule until ctx ends. A torrent limit a
// stopped controller left at the line-test value is put back first (R-1250), before any rule reads it.
func startUpwatch(ctx context.Context, svc *upwatch.Service, logger *log.Logger) {
	svc.RecoverHold(ctx)
	svc.RefreshMeasured(ctx)
	go svc.Watcher.Run(ctx, time.Minute)
	go func() {
		t := time.NewTicker(15 * time.Minute)
		defer t.Stop()
		for {
			select {
			case <-ctx.Done():
				return
			case <-t.C:
				svc.SeedLineTestNow(ctx) // R-1250: holds the torrent app only while the agent waits for quiet
				svc.QBLimitNow(ctx)
			}
		}
	}()
	l, ok := svc.Line()
	logger.Printf("[INFO] [upwatch] upload and CPU watch started (line %s, qBittorrent share %.0f %%)", lineText(l, ok), qbLimitShare*100)
}

func lineText(l float64, ok bool) string {
	if !ok {
		return "unknown — no weekly line test yet and no counting off-site push"
	}
	return upwatch.MbpsText(int64(l*1e6/8)) + " Mbit/s"
}