| index ed8e45cb2..29935be43 100644 |
| --- a/internal/console/economicspass.go |
| +++ b/internal/console/economicspass.go |
| @@ -505,13 +505,152 @@ func (s *Server) foldStorageAndCensus(ctx context.Context, from, to time.Time, d |
| return nil |
| } |
| |
| +// rollupPresenceHours converts one box's day rollup into billable box-hours |
| +// under the fleet-hours presence rule (task-3 brief): fleet-box rollups may |
| +// carry only the host-level aggregate (no per-unit CPU sampling — the box |
| +// runs no CARLOS-managed unit at all, just the remote-host-agent itself), so |
| +// presence is derived from the Host aggs rather than Units. For each PRESENT |
| +// hour (a key in d.Hours, regardless of what it carries): |
| +// - Host["cpu"].N minutes/60 when the host cpu agg is present; |
| +// - else the max N across whatever OTHER Host aggs the hour does carry; |
| +// - else — Host itself has no aggs at all for the hour — a flat 60-minute |
| +// (1.0h) fallback, since a present-but-metric-less hour key still means |
| +// the box was up and dialed in for that hour. |
| +func rollupPresenceHours(d metrics.DayRollup) float64 { |
| + var total float64 |
| + for _, hour := range d.Hours { |
| + if cpu, ok := hour.Host["cpu"]; ok { |
| + total += float64(cpu.N) / 60 |
| + continue |
| + } |
| + maxN := 0 |
| + for _, agg := range hour.Host { |
| + if agg.N > maxN { |
| + maxN = agg.N |
| + } |
| + } |
| + if maxN > 0 { |
| + total += float64(maxN) / 60 |
| + continue |
| + } |
| + total += 1.0 |
| + } |
| + return total |
| +} |
| + |
| +// fleetMeters derives per-account fleet box-hours for [from, to]: for every |
| +// box registered to every fleet (ALL statuses — a detached box's past hours |
| +// still bill, same posture as a deleted-account's already-written days), |
| +// account -> {fleet_box_hours_server, fleet_box_hours_device}, split by |
| +// whether the box's Tags carry the literal "device" tag. |
| +// |
| +// Deliberately a SECOND bounded read over metrics day objects, not an |
| +// extension of sweepUsage's single pass: sweepUsage iterates every box under |
| +// metrics/boxes/ once and attributes by UNIT (unitOwner), which has no |
| +// concept of a fleet box at all — a fleet box's day rollup carries no units |
| +// most of the time (see rollupPresenceHours's doc comment), only a host |
| +// aggregate, so folding this in there would mean threading a label->owner |
| +// map through sweepUsage's hot loop and its box-vs-fleet-box branching for a |
| +// meter family only two of usageSweep's several callers need at all. This |
| +// keeps sweepUsage and its existing tests completely untouched: the read |
| +// here is bounded to exactly the boxes that ARE fleet boxes (one Get per |
| +// fleet-box label per day in range; a day with no rollup is skipped, same |
| +// missing-means-skip convention sweepUsage itself uses). |
| +// |
| +// Label collisions across fleets — the registry enforces per-fleet box-label |
| +// uniqueness only, not a global one — resolve to the alphabetically-first |
| +// fleet name (ListFleets already returns fleets ordered by name, so the |
| +// first owner claimed for a label during that ordered walk wins) with one |
| +// log line per pass per colliding label. |
| +func (s *Server) fleetMeters(ctx context.Context, from, to time.Time) (map[string]map[string]float64, error) { |
| + reg, err := s.openPassRegistry() |
| + if err != nil { |
| + return nil, fmt.Errorf("console: fleet meters: %w", err) |
| + } |
| + fleets, err := reg.ListFleets() |
| + if err != nil { |
| + return nil, fmt.Errorf("console: fleet meters: %w", err) |
| + } |
| + |
| + type boxOwner struct { |
| + acct string |
| + device bool |
| + } |
| + labelOwner := map[string]boxOwner{} |
| + labelFleet := map[string]string{} // label -> fleet that claimed it, for the collision log line |
| + loggedCollision := map[string]bool{} |
| + |
| + for _, f := range fleets { |
| + boxes, err := reg.ListFleetBoxes(f.Name) |
| + if err != nil { |
| + return nil, fmt.Errorf("console: fleet meters: list boxes %s: %w", f.Name, err) |
| + } |
| + for _, b := range boxes { |
| + if _, exists := labelOwner[b.Label]; exists { |
| + if !loggedCollision[b.Label] { |
| + loggedCollision[b.Label] = true |
| + log.Printf("console: fleet meters: box label %q claimed by both fleet %q and fleet %q; %q (alphabetically first) wins", |
| + b.Label, labelFleet[b.Label], f.Name, labelFleet[b.Label]) |
| + } |
| + continue |
| + } |
| + device := false |
| + for _, tag := range strings.Split(b.Tags, ",") { |
| + if tag == "device" { |
| + device = true |
| + break |
| + } |
| + } |
| + labelOwner[b.Label] = boxOwner{acct: f.Account, device: device} |
| + labelFleet[b.Label] = f.Name |
| + } |
| + } |
| + |
| + out := map[string]map[string]float64{} |
| + touch := func(acct string) map[string]float64 { |
| + mm := out[acct] |
| + if mm == nil { |
| + mm = map[string]float64{} |
| + out[acct] = mm |
| + } |
| + return mm |
| + } |
| + |
| + fromDay := from.UTC().Truncate(24 * time.Hour) |
| + toDay := to.UTC().Truncate(24 * time.Hour) |
| + for label, owner := range labelOwner { |
| + meter := usage.MeterFleetBoxHoursServer |
| + if owner.device { |
| + meter = usage.MeterFleetBoxHoursDevice |
| + } |
| + for d := fromDay; !d.After(toDay); d = d.AddDate(0, 0, 1) { |
| + date := usage.DayOf(d) |
| + key := metrics.DayKey(label, date) |
| + raw, _, gerr := s.cfg.Releases.Get(ctx, key) |
| + if gerr != nil { |
| + continue // missing = skip, same convention as sweepUsage |
| + } |
| + var roll metrics.DayRollup |
| + if jerr := json.Unmarshal(raw, &roll); jerr != nil { |
| + log.Printf("console: fleet meters: parse %s: %v", key, jerr) |
| + continue |
| + } |
| + touch(owner.acct)[meter] += rollupPresenceHours(roll) |
| + } |
| + } |
| + return out, nil |
| +} |
| + |
| // statementsPass folds the month's metrics, edge and storage days (plus |
| -// census, for app_months) into one usage.Statement per account, create-or- |
| -// replace via usage.UpsertStatement — safe to run repeatedly as new days' |
| -// objects appear. It is a thin caller: open the registry (refusing early — |
| -// see openPassRegistry — since this is the WRITE pass), sweep the whole |
| -// month to date (sweepUsage), fold in storage/census meters |
| -// (foldStorageAndCensus), then upsert. |
| +// census, for app_months, and fleet box-hours, for fleet_box_hours_server/ |
| +// device) into one usage.Statement per account, create-or-replace via |
| +// usage.UpsertStatement — safe to run repeatedly as new days' objects |
| +// appear. It is a thin caller: open the registry (refusing early — see |
| +// openPassRegistry — since this is the WRITE pass), sweep the whole month to |
| +// date (sweepUsage), fold in storage/census meters (foldStorageAndCensus), |
| +// fold in fleet meters (fleetMeters — a union merge, since an account with |
| +// ONLY fleet usage and no platform routes at all must still get every meter |
| +// key it's owed), then upsert. |
| // |
| // Known shortfall, recorded rather than fixed at modelling grade: an account |
| // deleted mid-month drops out of s.store.Accounts() (its query filters |
| @@ -542,6 +681,20 @@ func (s *Server) statementsPass(ctx context.Context, now time.Time) error { |
| if err := s.foldStorageAndCensus(ctx, monthStart, nowUTC, meters); err != nil { |
| return fmt.Errorf("console: statements pass: %w", err) |
| } |
| + fleetMM, err := s.fleetMeters(ctx, monthStart, nowUTC) |
| + if err != nil { |
| + return fmt.Errorf("console: statements pass: %w", err) |
| + } |
| + for acct, mm := range fleetMM { |
| + dst := meters[acct] |
| + if dst == nil { |
| + dst = map[string]float64{} |
| + meters[acct] = dst |
| + } |
| + for meter, v := range mm { |
| + dst[meter] += v |
| + } |
| + } |
| coveredDays := sweep.CoveredDays |
| |
| // Every account gets a statement this pass, even one with zero usage — |
amadan