Skip to content

Commit bc87977

Browse files
hanthorscanner
andauthored
✨ feat(nous): close recovered observations after seven days (#12659)
* ✨ feat(nous): close recovered observations after seven days Track observation issue lifecycles on each spoke and relay recovery comments and closes through the hub. Add the durable ledger, tests, and operator docs. Hive-Run: #12655 Hive-Plan: #12341 child 4 — observation recovery lifecycle Hive-Spec: #12655#acceptance-criteria Signed-off-by: hanthor <jreilly1821@gmail.com> * test(nous): count the recovery window from the first inactive cycle The ledger starts the seven-day recovery window on the first eval cycle that no longer reports the finding, so a single reconcile eight days after the last active sample cannot produce a recovery request. The three failing tests assumed the window counted from the last active sample; align them with the documented ledger semantics. Signed-off-by: scanner <scanner@hive.kubestellar.io> * test(nous): add coverage for observation ledger methods and error paths Signed-off-by: hanthor <jreilly1821@gmail.com> --------- Signed-off-by: hanthor <jreilly1821@gmail.com> Signed-off-by: scanner <scanner@hive.kubestellar.io> Co-authored-by: scanner <scanner@hive.kubestellar.io>
1 parent acad67b commit bc87977

8 files changed

Lines changed: 1285 additions & 32 deletions

File tree

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
- Add Nous observation recovery tracking with a seven-day lifecycle, hub-relayed recovery comments, and issue closure.

‎src/cmd/hive/selfreport_publish.go‎

Lines changed: 126 additions & 24 deletions
Original file line numberDiff line numberDiff line change
@@ -20,6 +20,7 @@ import (
2020
"github.com/hivecommons/hive/pkg/github"
2121
"github.com/hivecommons/hive/pkg/hivestate"
2222
"github.com/hivecommons/hive/pkg/hub/spoke"
23+
"github.com/hivecommons/hive/pkg/nous"
2324
)
2425

2526
// Self-report publisher (hivecommons/hive#11868). Each governor cycle it takes
@@ -28,7 +29,7 @@ import (
2829
//
2930
// - dry-run → nothing (preview stays on the dashboard);
3031
// - Nous observations → the hub relay only; the spoke GitHub client is never used;
31-
// - the API budget is not normal → defer ordinary reports (the report re-emits next cycle);
32+
// - the API budget is not normal → defer ordinary reports (observations still relay);
3233
// - fingerprint already written by this hive → at most one "also seen"
3334
// comment per 24h, otherwise skip;
3435
// - fingerprint closed upstream inside the 7-day cooldown → record and skip
@@ -44,6 +45,7 @@ import (
4445

4546
// selfReportHubRelayPath is the hub ingest the spoke falls back to.
4647
const selfReportHubRelayPath = "/api/selfreport/ingest"
48+
const selfReportObservationClass = "observation"
4749

4850
// selfReportRelayTimeout bounds one relay POST.
4951
const selfReportRelayTimeout = 20 * time.Second
@@ -62,10 +64,11 @@ type selfReportRelay func(ctx context.Context, req selfReportRelayRequest) (self
6264

6365
// selfReportRelayRequest is the wire shape shared with pkg/hub.
6466
type selfReportRelayRequest struct {
67+
Action string `json:"action,omitempty"`
6568
HiveID string `json:"hive_id"`
6669
Fingerprint string `json:"fingerprint"`
6770
Class string `json:"class,omitempty"`
68-
Title string `json:"title"`
71+
Title string `json:"title,omitempty"`
6972
Body string `json:"body"`
7073
Labels []string `json:"labels,omitempty"`
7174
Version string `json:"version,omitempty"`
@@ -81,33 +84,36 @@ type selfReportRelayResponse struct {
8184
Number int `json:"number"`
8285
URL string `json:"url"`
8386
Joined bool `json:"joined"`
87+
Reason string `json:"reason,omitempty"`
8488
}
8589

8690
type selfReportPublishDeps struct {
87-
dryRun bool
88-
gh selfReportWriter
89-
relay selfReportRelay
90-
repo string
91-
hiveSlug string
92-
version string
93-
cluster string
94-
limits fleetreport.FilingLimits
95-
now func() time.Time
96-
tracker *fleetreport.RuntimeTracker
97-
budgetOK func() bool
98-
hubLinked bool
91+
dryRun bool
92+
gh selfReportWriter
93+
relay selfReportRelay
94+
repo string
95+
hiveSlug string
96+
version string
97+
cluster string
98+
limits fleetreport.FilingLimits
99+
now func() time.Time
100+
tracker *fleetreport.RuntimeTracker
101+
budgetOK func() bool
102+
hubLinked bool
103+
observations *nous.ObservationLedger
99104
}
100105

101106
func newSelfReportPublishDeps(cfg *config.Config, ghClient *github.Client, dashSrv *dashboard.Server) selfReportPublishDeps {
102107
version, _ := dashboardBuildInfo()
103108
d := selfReportPublishDeps{
104-
dryRun: true,
105-
repo: config.DefaultFleetReportRepo,
106-
limits: fleetreport.DefaultFilingLimits,
107-
now: time.Now,
108-
tracker: fleetreport.DefaultTracker(),
109-
version: version,
110-
cluster: os.Getenv("HIVE_CLUSTER_ID"),
109+
dryRun: true,
110+
repo: config.DefaultFleetReportRepo,
111+
limits: fleetreport.DefaultFilingLimits,
112+
now: time.Now,
113+
tracker: fleetreport.DefaultTracker(),
114+
version: version,
115+
cluster: os.Getenv("HIVE_CLUSTER_ID"),
116+
observations: nous.NewObservationLedger(nous.DefaultObservationLedgerPath),
111117
}
112118
if ghClient != nil {
113119
d.gh = ghClient
@@ -158,12 +164,44 @@ func publishFleetReportsWith(ctx context.Context, logger *slog.Logger, dashSrv *
158164
}
159165
now := deps.now()
160166
state := dashSrv.FleetReportState()
167+
var observationRecoveries []nous.RecoveryRequest
168+
if deps.observations != nil {
169+
active := make([]nous.ActiveObservation, 0)
170+
for _, report := range res.Reports {
171+
if !isObservationReport(report) {
172+
continue
173+
}
174+
before, after := observationBeforeAfter(report)
175+
open := state.Open[report.Fingerprint]
176+
issueNumber, issueURL := open.Number, open.URL
177+
if open.Recovered || !open.ClosedAt.IsZero() {
178+
issueNumber, issueURL = 0, ""
179+
}
180+
active = append(active, nous.ActiveObservation{
181+
Fingerprint: report.Fingerprint,
182+
IssueNumber: issueNumber,
183+
URL: issueURL,
184+
Before: before,
185+
After: after,
186+
Evidence: report.Body,
187+
})
188+
}
189+
var err error
190+
observationRecoveries, err = deps.observations.Reconcile(active, now)
191+
if err != nil {
192+
logger.Warn("fleet report: observation lifecycle ledger failed", "error", err)
193+
}
194+
}
161195
for _, report := range res.Reports {
162-
if !isObservationReport(report) && !budgetNormal {
196+
isObservation := isObservationReport(report)
197+
if !isObservation && !budgetNormal {
163198
logger.Info("fleet report: API budget not normal; deferring non-observation self-report", "fingerprint", report.Fingerprint)
164199
continue
165200
}
166201
open := state.Open[report.Fingerprint]
202+
if isObservation && open.Recovered {
203+
open = fleetreport.OpenIssue{}
204+
}
167205
if fleetreport.ClosedRecently(open.ClosedAt, now, deps.limits) {
168206
continue
169207
}
@@ -174,7 +212,7 @@ func publishFleetReportsWith(ctx context.Context, logger *slog.Logger, dashSrv *
174212

175213
var write github.FleetReportResult
176214
var err error
177-
if isObservationReport(report) {
215+
if isObservation {
178216
// Observations are hub-owned reports. Never fall back to the
179217
// spoke's GitHub client, even when it can write another repo.
180218
if deps.relay == nil {
@@ -246,6 +284,9 @@ func publishFleetReportsWith(ctx context.Context, logger *slog.Logger, dashSrv *
246284
created := write.Created
247285
dashSrv.UpdateFleetReportState(func(st *fleetreport.State) {
248286
o := st.Open[report.Fingerprint]
287+
if isObservation && o.Recovered {
288+
o = fleetreport.OpenIssue{}
289+
}
249290
o.Number, o.URL = write.Number, write.URL
250291
o.OpenedByHive = o.OpenedByHive || created
251292
o.Joined = o.Joined || write.Joined
@@ -260,11 +301,40 @@ func publishFleetReportsWith(ctx context.Context, logger *slog.Logger, dashSrv *
260301
fleetreport.RecordFiling(st, now)
261302
}
262303
})
304+
if isObservationReport(report) && deps.observations != nil && write.Number > 0 {
305+
if err := deps.observations.RecordIssue(report.Fingerprint, write.Number, write.URL, now); err != nil {
306+
logger.Warn("fleet report: observation issue ledger update failed", "fingerprint", report.Fingerprint, "error", err)
307+
}
308+
}
263309
state = dashSrv.FleetReportState()
264310
logger.Info("fleet report: upstream report recorded", "fingerprint", report.Fingerprint, "class", report.Class, "issue", write.Number, "created", created, "joined", write.Joined, "commented", write.Commented, "body_updated", write.BodyUpdated)
265311
}
266312

313+
for _, recovery := range observationRecoveries {
314+
report := fleetreport.Report{
315+
Fingerprint: recovery.Fingerprint,
316+
Trigger: selfReportObservationClass,
317+
Class: selfReportObservationClass,
318+
Body: recovery.Comment,
319+
}
320+
if _, err := relayObservationRecovery(ctx, deps, report); err != nil {
321+
logger.Warn("fleet report: observation recovery relay failed", "fingerprint", recovery.Fingerprint, "issue", recovery.IssueNumber, "error", err)
322+
continue
323+
}
324+
if deps.observations != nil {
325+
if err := deps.observations.MarkClosed(recovery.Fingerprint, now); err != nil {
326+
logger.Warn("fleet report: observation close ledger update failed", "fingerprint", recovery.Fingerprint, "error", err)
327+
continue
328+
}
329+
}
330+
dashSrv.MarkFleetReportRecovered(recovery.Fingerprint)
331+
logger.Info("fleet report: observation recovery recorded", "fingerprint", recovery.Fingerprint, "issue", recovery.IssueNumber)
332+
}
267333
for _, report := range res.Recoveries {
334+
if isObservationReport(report) {
335+
// Observation recovery is gated by the durable seven-day ledger above.
336+
continue
337+
}
268338
open, ok := state.Open[report.Fingerprint]
269339
if !ok {
270340
continue
@@ -346,6 +416,38 @@ func firstEvidence(report fleetreport.Report) fleetreport.Evidence {
346416
return fleetreport.Evidence{}
347417
}
348418

419+
func observationBeforeAfter(report fleetreport.Report) (string, string) {
420+
const seriesPrefix = "| series | `"
421+
start := strings.Index(report.Body, seriesPrefix)
422+
if start < 0 {
423+
return "", ""
424+
}
425+
seriesStart := start + len(seriesPrefix)
426+
seriesEnd := strings.IndexByte(report.Body[seriesStart:], '`')
427+
if seriesEnd < 0 {
428+
return "", ""
429+
}
430+
var series []float64
431+
if err := json.Unmarshal([]byte(report.Body[seriesStart:seriesStart+seriesEnd]), &series); err != nil || len(series) < 2 {
432+
return "", ""
433+
}
434+
return fmt.Sprintf("%g", series[0]), fmt.Sprintf("%g", series[len(series)-1])
435+
}
436+
437+
func relayObservationRecovery(ctx context.Context, deps selfReportPublishDeps, report fleetreport.Report) (selfReportRelayResponse, error) {
438+
if deps.relay == nil {
439+
return selfReportRelayResponse{}, errSelfReportObservationNoRelay
440+
}
441+
return deps.relay(ctx, selfReportRelayRequest{
442+
Action: "recover",
443+
HiveID: deps.hiveSlug,
444+
Fingerprint: report.Fingerprint,
445+
Class: selfReportObservationClass,
446+
Body: report.Body,
447+
Repo: config.DefaultFleetReportRepo,
448+
})
449+
}
450+
349451
// selfReportShouldRelay: a 403/404 from the report repo means this hive's
350452
// credential cannot write there (self-hosted PAT scoped to its own repos, or
351453
// the App not installed on the report org). Anything else is a real error.
@@ -394,7 +496,7 @@ func hubSelfReportRelay(hubURL string, bearer func() string) selfReportRelay {
394496
if err := json.Unmarshal(raw, &out); err != nil {
395497
return selfReportRelayResponse{}, err
396498
}
397-
if !out.OK {
499+
if !out.OK && req.Action != "recover" {
398500
return selfReportRelayResponse{}, errors.New("hub declined the self-report")
399501
}
400502
return out, nil

‎src/cmd/hive/selfreport_publish_test.go‎

Lines changed: 96 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@ import (
1515
"github.com/hivecommons/hive/pkg/dashboard"
1616
"github.com/hivecommons/hive/pkg/fleetreport"
1717
"github.com/hivecommons/hive/pkg/github"
18+
"github.com/hivecommons/hive/pkg/nous"
1819
)
1920

2021
// fakeSelfReportWriter scripts the GitHub seam the publisher drives.
@@ -273,3 +274,98 @@ func TestSelfReportShouldRelay(t *testing.T) {
273274
}
274275
}
275276
}
277+
278+
func TestPublishObservationRecoveryUnknownHubIssueIsNoop(t *testing.T) {
279+
srv := selfReportTestServer(t)
280+
now := time.Date(2026, 9, 24, 12, 0, 0, 0, time.UTC)
281+
fp := "observation-recovery"
282+
ledger := nous.NewObservationLedger(filepath.Join(t.TempDir(), "observations.json"))
283+
if _, err := ledger.Reconcile([]nous.ActiveObservation{{
284+
Fingerprint: fp,
285+
IssueNumber: 42,
286+
URL: "https://github.com/hivecommons/hive/issues/42",
287+
Before: "p90=900ms",
288+
After: "p90=100ms",
289+
Evidence: "last sample",
290+
}}, now); err != nil {
291+
t.Fatal(err)
292+
}
293+
if err := ledger.RecordIssue(fp, 42, "https://github.com/hivecommons/hive/issues/42", now); err != nil {
294+
t.Fatal(err)
295+
}
296+
srv.UpdateFleetReportState(func(st *fleetreport.State) {
297+
st.Open[fp] = fleetreport.OpenIssue{
298+
Number: 42, URL: "https://github.com/hivecommons/hive/issues/42",
299+
Trigger: selfReportObservationClass, Class: selfReportObservationClass, PostedAt: now,
300+
}
301+
})
302+
w := &fakeSelfReportWriter{}
303+
var relayed []selfReportRelayRequest
304+
deps := selfReportDeps(w, &now)
305+
deps.observations = ledger
306+
deps.relay = func(_ context.Context, req selfReportRelayRequest) (selfReportRelayResponse, error) {
307+
relayed = append(relayed, req)
308+
return selfReportRelayResponse{Reason: "unknown_fingerprint"}, nil
309+
}
310+
// The first cycle without the finding starts the seven-day window; the
311+
// recovery request is produced only by a later cycle past that window.
312+
publishFleetReportsWith(context.Background(), slog.Default(), srv, &fleetreport.Result{}, deps)
313+
if len(relayed) != 0 {
314+
t.Fatalf("recovery relayed before the seven-day window: %#v", relayed)
315+
}
316+
now = now.Add(8 * 24 * time.Hour)
317+
publishFleetReportsWith(context.Background(), slog.Default(), srv, &fleetreport.Result{}, deps)
318+
if len(relayed) != 1 {
319+
t.Fatalf("recovery relays = %d, want 1", len(relayed))
320+
}
321+
if relayed[0].Action != "recover" || relayed[0].Class != selfReportObservationClass || !strings.Contains(relayed[0].Body, "p90=900ms") || !strings.Contains(relayed[0].Body, "p90=100ms") {
322+
t.Fatalf("recovery relay = %#v", relayed[0])
323+
}
324+
records, err := ledger.Records()
325+
if err != nil {
326+
t.Fatal(err)
327+
}
328+
if records[fp].ClosedAt.IsZero() {
329+
t.Fatalf("unknown hub fingerprint was not acknowledged locally: %#v", records[fp])
330+
}
331+
if !srv.FleetReportState().Open[fp].Recovered {
332+
t.Fatal("dashboard state did not record observation recovery")
333+
}
334+
}
335+
336+
func TestPublishObservationReportsRelayThroughHubOnly(t *testing.T) {
337+
srv := selfReportTestServer(t)
338+
now := time.Date(2026, 9, 24, 12, 0, 0, 0, time.UTC)
339+
fp := "h1:queue-growth"
340+
ledger := nous.NewObservationLedger(filepath.Join(t.TempDir(), "observations.json"))
341+
w := &fakeSelfReportWriter{ensure: func(fleetreport.Report, github.FleetReportOptions) (github.FleetReportResult, error) {
342+
t.Fatal("observation used the spoke GitHub writer")
343+
return github.FleetReportResult{}, nil
344+
}}
345+
var relayed []selfReportRelayRequest
346+
deps := selfReportDeps(w, &now)
347+
deps.observations = ledger
348+
deps.budgetOK = func() bool { return false }
349+
deps.relay = func(_ context.Context, req selfReportRelayRequest) (selfReportRelayResponse, error) {
350+
relayed = append(relayed, req)
351+
return selfReportRelayResponse{OK: true, Number: 7, URL: "https://github.com/hivecommons/hive/issues/7"}, nil
352+
}
353+
report := fleetreport.Report{
354+
Fingerprint: fp,
355+
Trigger: selfReportObservationClass,
356+
Class: selfReportObservationClass,
357+
Title: "Queue depth is growing",
358+
Body: "<!-- hive-fleet-fingerprint:h1:queue-growth -->\n| series | `[10,20]` |",
359+
}
360+
publishFleetReportsWith(context.Background(), slog.Default(), srv, &fleetreport.Result{Reports: []fleetreport.Report{report}}, deps)
361+
if len(relayed) != 1 || relayed[0].Repo != "hivecommons/hive" || relayed[0].Action != "" || relayed[0].Class != selfReportObservationClass {
362+
t.Fatalf("observation relay = %#v", relayed)
363+
}
364+
records, err := ledger.Records()
365+
if err != nil {
366+
t.Fatal(err)
367+
}
368+
if records[fp].IssueNumber != 7 || records[fp].Before != "10" || records[fp].After != "20" {
369+
t.Fatalf("observation ledger = %#v", records[fp])
370+
}
371+
}

‎src/docs/strategy-lab.md‎

Lines changed: 22 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -40,6 +40,28 @@ Nous state lives in three places, none of which is `hive.yaml`:
4040
| Experiment campaigns | `/data/nous/campaigns.json` | `pkg/nous` campaign store (file version 2), written by `POST/PATCH /api/nous/campaigns` (including restarts) and by the legacy `PUT /api/nous/config/goals` / `POST /api/nous/campaign/start`; see [Campaigns and the per-campaign baseline](#campaigns-and-the-per-campaign-baseline) |
4141
| Mode, output mode, repo allow-list, fast-fail bounds, schedule | in-memory `NousState.Config`, set via `PUT /api/nous/config/*` | dashboard ⚙ *Experiment Lab Configuration* dialog (lab-wide guardrails only) or direct API calls |
4242

43+
### Observation finding recovery
44+
45+
Observation detectors keep their issue lifecycle in
46+
`/data/nous/observations.json`, keyed by finding fingerprint. Each record stores
47+
the issue number and URL, `opened_at`, `last_seen_at`, and (while the finding is
48+
absent) `recovered_since`, plus the before/after values and last evidence used
49+
in the recovery comment.
50+
51+
Every evaluation reconciles the currently active findings with this ledger. An
52+
active finding clears `recovered_since`; an absent finding starts the recovery
53+
window. After **seven consecutive days** without the finding, the spoke sends
54+
the hub relay an `action: "recover"` request. The hub searches for the open
55+
issue marker, comments the before/after values and evidence, and closes the
56+
issue. A missing or already-closed fingerprint is an idempotent `ok: false`
57+
no-op, not an evaluation error.
58+
59+
After a successful recovery, the ledger records `closed_at`. If the same
60+
fingerprint fires again, it starts a fresh lifecycle and the hub's open-only
61+
search creates a new issue rather than reopening the old one. The spoke never
62+
writes recovery comments or closes issues directly; the hub relay is the
63+
authoritative write path.
64+
4365
The in-memory `Config` map (goals/repos/output/fast_fail/schedule/
4466
principles sections) is **not currently persisted back to disk
4567
across a process restart** — it is rebuilt from defaults each time the hive

0 commit comments

Comments
 (0)