kami/Maven
1
0
Fork 0
Code Issues 1 Pull Requests 1 Actions Packages Projects Releases Wiki Activity

sweep capture and mavpoll for duplication and magic values (V-581)

Browse Source 
capture: one Session.discard for the reaper and Abort, which both closed
the spool and deleted it by hand. One abandon closure for the two failed
Start paths. bytesPerSample and bytesPerSecond replace three copies of
the byte-rate arithmetic.

mavpoll: unchanged and writeFact hold the read-compare and the row shape,
so the three fact writers keep only what differs between them, their log
line. The wg key and source are named constants. Named coreDialWait,
zenTimeoutFactor, maxBodyBytes and factConfidence. run lost the zenmoney
setup to newZenClient and the ticker to loop.

Comments only where they were stale: the package doc counted two sources
and there are four.
This commit is contained in:
claude
2026-08-06 01:54:38 +04:00
parent 59b98d5c4d
commit db8cbdc20a
2 changed files with 144 additions and 84 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/mavpoll/main.go
+106 -58
View File
@@ -6,10 +6,11 @@
// restart-free, fail-independent — a crashing poller can't touch the store key
// (it never had it), worst case a stale env fact until the next tick.
//
// Two sources, each its own provenance (the loop's rules trust source):
// Four sources, each its own provenance (the loop's rules trust source):
// - netdata → poll:netdata resource alarms (disk/mem/cert/temp)
// - kuma → poll:uptimekuma service up/down (the source of truth for it)
// - zenmoney → poll:zenmoney spending/income totals (Vikunja #125)
// - wireguard → infer:wg latest handshake, the presence signal
//
// The zenmoney source is why the token lives HERE and not in core: the poller
// already owns every other third-party credential, it holds no store key, and
@@ -77,26 +78,15 @@ func run(args []string) error {
return fmt.Errorf("nothing to poll: set -netdata, -kuma, -wg and/or -zenmoney-token-file")
}
// The token is read from a file, never taken as a flag value: an argv token
// is visible in `ps` to every user on the box and lands in the compose file
// and the shell history. Read once at start — a rotated token means a
// restart, which is cheaper than re-reading his credential every hour.
var zen *zenmoney.Client
if *zenTokenFile != "" {
raw, err := os.ReadFile(*zenTokenFile)
if err != nil {
return fmt.Errorf("read zenmoney token: %w", err)
}
zen, err = zenmoney.New(strings.TrimSpace(string(raw)), *zenURL, *timeout*3)
if err != nil {
return err
}
zen, err := newZenClient(*zenTokenFile, *zenURL, *timeout)
if err != nil {
return err
}
ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
defer stop()
core, err := ipc.DialWait(*socket, 60*time.Second)
core, err := ipc.DialWait(*socket, coreDialWait)
if err != nil {
return err
}
@@ -117,14 +107,46 @@ func run(args []string) error {
// The token is never logged, not even its length.
log.Printf("mavpoll: polling every %s (netdata=%q kuma=%q wg=%q zenmoney=%v every %s)",
*interval, *netdataURL, *kumaURL, *wgIface, zen != nil, *zenInterval)
p.loop(ctx, *interval)
return nil
}
// coreDialWait — how long to wait for core's socket at start. The poller and
// core come up together under compose, so a cold start is a wait, not a failure.
const coreDialWait = 60 * time.Second
// zenTimeoutFactor — the zenmoney client gets a longer deadline than the other
// sources. A diff call walks his whole transaction history, where netdata and
// kuma answer from memory.
const zenTimeoutFactor = 3
// newZenClient builds the money client, or nil when no token file was given.
//
// The token is read from a file, never taken as a flag value: an argv token is
// visible in `ps` to every user on the box and lands in the compose file and
// the shell history. Read once at start — a rotated token means a restart,
// which is cheaper than re-reading his credential every hour.
func newZenClient(tokenFile, baseURL string, timeout time.Duration) (*zenmoney.Client, error) {
if tokenFile == "" {
return nil, nil
}
raw, err := os.ReadFile(tokenFile)
if err != nil {
return nil, fmt.Errorf("read zenmoney token: %w", err)
}
return zenmoney.New(strings.TrimSpace(string(raw)), baseURL, timeout*zenTimeoutFactor)
}
// loop polls until the context is cancelled.
func (p *poller) loop(ctx context.Context, interval time.Duration) {
p.pollOnce(ctx) // fire immediately; don't idle a full interval on start
t := time.NewTicker(*interval)
t := time.NewTicker(interval)
defer t.Stop()
for {
select {
case <-ctx.Done():
log.Printf("mavpoll: bye")
return nil
return
case <-t.C:
p.pollOnce(ctx)
}
@@ -151,8 +173,8 @@ type poller struct {
zenLast time.Time
}
// pollOnce — one sweep of both sources. A failure in one source logs and does
// NOT abort the other: netdata being down shouldn't blind kuma and vice versa.
// pollOnce — one sweep of every configured source. A failure in one logs and
// does NOT abort the rest: netdata being down shouldn't blind kuma.
func (p *poller) pollOnce(ctx context.Context) {
now := time.Now()
if p.netdataURL != "" {
@@ -244,6 +266,14 @@ func (p *poller) pollZenmoney(ctx context.Context, now time.Time) error {
// ---- wireguard: latest handshake → presence signal -------------------------
const (
// wgFactKey / wgSource — the presence signal, read by the decay in core.
// The source says infer because a handshake is evidence he is home, not a
// reading of where he is.
wgFactKey = "wg_handshake"
wgSource = "infer:wg"
)
// pollWg reads `wg show <iface> latest-handshakes` and writes a wg_handshake
// fact (source=infer:wg) stamped with the MOST RECENT peer handshake time — not
// now(). Presence decays from the real handshake instant, so the fact's ts must
@@ -263,20 +293,19 @@ func (p *poller) pollWg(ctx context.Context) error {
return nil // no peer has ever handshaked → drop out of presence
}
hs := time.Unix(maxTs, 0)
prev, err := p.core.LatestFactBySource(ctx, "wg_handshake", "infer:wg")
prev, err := p.core.LatestFactBySource(ctx, wgFactKey, wgSource)
if err == nil && !hs.After(prev.Ts) {
return nil // not newer → no churn
}
if err != nil && err != ipc.ErrNoFact && !isNoFact(err) {
return fmt.Errorf("read wg_handshake: %w", err)
if err != nil && !isNoFact(err) {
return fmt.Errorf("read %s: %w", wgFactKey, err)
}
if _, err := p.core.WriteFact(ctx, ipc.WriteFactReq{
Ts: hs, Kind: "env", Key: "wg_handshake", Value: `"up"`,
Source: "infer:wg", Confidence: 1.0,
}); err != nil {
return fmt.Errorf("write wg_handshake: %w", err)
// The ts is the handshake instant, not now(): presence decays from when he
// was last seen.
if err := p.writeFact(ctx, wgFactKey, wgSource, `"up"`, hs); err != nil {
return err
}
log.Printf("mavpoll: wg_handshake @ %s (infer:wg)", hs.Format(time.RFC3339))
log.Printf("mavpoll: %s @ %s (%s)", wgFactKey, hs.Format(time.RFC3339), wgSource)
return nil
}
@@ -431,29 +460,53 @@ func kumaState(v float64) string {
// ---- helpers ---------------------------------------------------------------
// writeIfChanged writes a `facts(kind=env)` row only when val differs from the
// latest fact for (key, source). Values are stored JSON-encoded (the store's
// convention: `"down"`, `"critical"`), matching how rules compare f.Value.
func (p *poller) writeIfChanged(ctx context.Context, key, source, val string, now time.Time) error {
jv, _ := json.Marshal(val) // string never fails to marshal
// factConfidence — every poll is a direct reading of another service, never an
// inference, so the fact goes in at full confidence.
const factConfidence = 1.0
// unchanged reports whether the latest fact for (key, source) already holds
// jsonVal. A missing fact is not an error here, it is the first write.
func (p *poller) unchanged(ctx context.Context, key, source, jsonVal string) (bool, error) {
prev, err := p.core.LatestFactBySource(ctx, key, source)
switch {
case err == nil && prev.Value == string(jv):
return nil // unchanged → no churn
case err != nil && err != ipc.ErrNoFact && !isNoFact(err):
return fmt.Errorf("read %s: %w", key, err)
case err == nil:
return prev.Value == jsonVal, nil
case isNoFact(err):
return false, nil
default:
return false, fmt.Errorf("read %s: %w", key, err)
}
_, err = p.core.WriteFact(ctx, ipc.WriteFactReq{
}
// writeFact writes one `facts(kind=env)` row. Every poll in this file lands
// here, so the row shape is written once.
func (p *poller) writeFact(ctx context.Context, key, source, jsonVal string, now time.Time) error {
_, err := p.core.WriteFact(ctx, ipc.WriteFactReq{
Ts: now,
Kind: "env",
Key: key,
Value: string(jv),
Value: jsonVal,
Source: source,
Confidence: 1.0, // a direct reading, not an inference
Confidence: factConfidence,
})
if err != nil {
return fmt.Errorf("write %s: %w", key, err)
}
return nil
}
// writeIfChanged writes only when val differs from the latest fact for
// (key, source). Values are stored JSON-encoded (the store's convention:
// `"down"`, `"critical"`), matching how rules compare f.Value.
func (p *poller) writeIfChanged(ctx context.Context, key, source, val string, now time.Time) error {
jv, _ := json.Marshal(val) // a string never fails to marshal
same, err := p.unchanged(ctx, key, source, string(jv))
if err != nil || same {
return err // unchanged → no churn
}
if err := p.writeFact(ctx, key, source, string(jv), now); err != nil {
return err
}
log.Printf("mavpoll: %s=%s (%s)", key, val, source)
return nil
}
@@ -466,18 +519,12 @@ func (p *poller) writeIfChanged(ctx context.Context, key, source, val string, no
// The log line names the key and the source, never the figures: mavpoll's log
// is not the place his spending ends up.
func (p *poller) writeIfChangedRaw(ctx context.Context, key, source, jsonVal string, now time.Time) error {
prev, err := p.core.LatestFactBySource(ctx, key, source)
switch {
case err == nil && prev.Value == jsonVal:
return nil
case err != nil && err != ipc.ErrNoFact && !isNoFact(err):
return fmt.Errorf("read %s: %w", key, err)
same, err := p.unchanged(ctx, key, source, jsonVal)
if err != nil || same {
return err
}
if _, err := p.core.WriteFact(ctx, ipc.WriteFactReq{
Ts: now, Kind: "env", Key: key, Value: jsonVal,
Source: source, Confidence: 1.0,
}); err != nil {
return fmt.Errorf("write %s: %w", key, err)
if err := p.writeFact(ctx, key, source, jsonVal, now); err != nil {
return err
}
log.Printf("mavpoll: %s updated (%s)", key, source)
return nil
@@ -490,11 +537,8 @@ func (p *poller) writeIfChangedRaw(ctx context.Context, key, source, jsonVal str
// The log line names the key only, never the figures: mavpoll's log is not the
// place his spending ends up.
func (p *poller) writeMoneyFact(ctx context.Context, key, jsonVal string, now time.Time) error {
if _, err := p.core.WriteFact(ctx, ipc.WriteFactReq{
Ts: now, Kind: "env", Key: key, Value: jsonVal,
Source: zenmoney.Source, Confidence: 1.0,
}); err != nil {
return fmt.Errorf("write %s: %w", key, err)
if err := p.writeFact(ctx, key, zenmoney.Source, jsonVal, now); err != nil {
return err
}
log.Printf("mavpoll: %s read (%s)", key, zenmoney.Source)
return nil
@@ -506,6 +550,10 @@ func isNoFact(err error) bool {
return errors.Is(err, ipc.ErrNoFact)
}
// maxBodyBytes caps what a source can make the poller hold. Kuma's whole
// metrics page is a few hundred kilobytes, so 4 MiB is slack, not a budget.
const maxBodyBytes = 4 << 20
func (p *poller) get(ctx context.Context, url, basicUser string) ([]byte, error) {
req, err := http.NewRequestWithContext(ctx, http.MethodGet, url, nil)
if err != nil {
@@ -519,7 +567,7 @@ func (p *poller) get(ctx context.Context, url, basicUser string) ([]byte, error)
return nil, err
}
defer resp.Body.Close()
body, err := io.ReadAll(io.LimitReader(resp.Body, 4<<20))
body, err := io.ReadAll(io.LimitReader(resp.Body, maxBodyBytes))
if err != nil {
return nil, err
}