From aaf0dba8fc42d34fd1a8e3e1bf2ed9379221f3a6 Mon Sep 17 00:00:00 2001 From: CMGS Date: Fri, 30 Oct 2026 22:45:23 +0800 Subject: [PATCH 01/16] usage: move the usage event type out of pool The shipper that lands the journal in the shared meta store decodes the same lines, so the line schema gets one owner both packages import. --- sandboxd/pool/archive.go | 5 +++-- sandboxd/pool/checkpoint.go | 3 ++- sandboxd/pool/claim.go | 11 ++++++----- sandboxd/pool/egress.go | 3 ++- sandboxd/pool/failover.go | 3 ++- sandboxd/pool/fork.go | 3 ++- sandboxd/pool/hibernate.go | 5 +++-- sandboxd/pool/journal.go | 19 ------------------- sandboxd/pool/liveness.go | 7 ++++--- sandboxd/pool/pool.go | 5 +++-- sandboxd/pool/telemetry.go | 5 +++-- sandboxd/pool/telemetry_test.go | 13 +++++++------ sandboxd/pool/template.go | 3 ++- sandboxd/pool/volume_rw_test.go | 3 ++- sandboxd/pool/volume_test.go | 3 ++- sandboxd/usage/usage.go | 25 +++++++++++++++++++++++++ 16 files changed, 68 insertions(+), 48 deletions(-) create mode 100644 sandboxd/usage/usage.go diff --git a/sandboxd/pool/archive.go b/sandboxd/pool/archive.go index e551bdfe6..2b379f680 100644 --- a/sandboxd/pool/archive.go +++ b/sandboxd/pool/archive.go @@ -14,6 +14,7 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/store" "github.com/cocoonstack/sandbox/sandboxd/types" + "github.com/cocoonstack/sandbox/sandboxd/usage" "github.com/cocoonstack/sandbox/sandboxd/utils" ) @@ -175,7 +176,7 @@ func (m *Manager) archive(ctx context.Context, sb *types.Sandbox) error { m.deleteOrphanArchiveCk(ctx, replica) } m.counters.archives.Add(1) - m.recordUsage(ctx, usageEvent{Event: "archive", ID: sb.ID, Reference: ck.ID, Tenant: sb.Tenant}) + m.recordUsage(ctx, usage.Event{Event: "archive", ID: sb.ID, Reference: ck.ID, Tenant: sb.Tenant}) return nil } @@ -244,7 +245,7 @@ func (m *Manager) wakeArchived(ctx context.Context, sb *types.Sandbox) (string, return "", ErrUnknownSandbox } m.counters.unarchives.Add(1) - m.recordUsage(ctx, usageEvent{Event: "unarchive", ID: sb.ID, Reference: ck, Tenant: sb.Tenant}) + m.recordUsage(ctx, usage.Event{Event: "unarchive", ID: sb.ID, Reference: ck, Tenant: sb.Tenant}) return built.VsockSocket, nil } diff --git a/sandboxd/pool/checkpoint.go b/sandboxd/pool/checkpoint.go index aed5f7c13..582d3c69b 100644 --- a/sandboxd/pool/checkpoint.go +++ b/sandboxd/pool/checkpoint.go @@ -15,6 +15,7 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/store" "github.com/cocoonstack/sandbox/sandboxd/types" + "github.com/cocoonstack/sandbox/sandboxd/usage" ) // DeleteScope selects how far a checkpoint delete reaches; fleet-wide is the zero value. @@ -53,7 +54,7 @@ func (m *Manager) Checkpoint(ctx context.Context, id string, cred Cred, name, te return types.Checkpoint{}, err } m.counters.checkpoints.Add(1) - m.recordUsage(ctx, usageEvent{Event: "checkpoint", ID: sb.ID, VMName: lockedVMName(sb), Reference: ckpt.ID}) + m.recordUsage(ctx, usage.Event{Event: "checkpoint", ID: sb.ID, VMName: lockedVMName(sb), Reference: ckpt.ID}) return ckpt, nil } diff --git a/sandboxd/pool/claim.go b/sandboxd/pool/claim.go index c13e8e75f..519e22f79 100644 --- a/sandboxd/pool/claim.go +++ b/sandboxd/pool/claim.go @@ -13,6 +13,7 @@ import ( "github.com/projecteru2/core/log" "github.com/cocoonstack/sandbox/sandboxd/types" + "github.com/cocoonstack/sandbox/sandboxd/usage" ) const ( @@ -357,7 +358,7 @@ func (m *Manager) releaseResolved(ctx context.Context, id string, sb *types.Sand m.out.Disarm(id, removed) m.dropSnap(ctx, snap) m.counters.releases.Add(1) - m.recordUsage(ctx, usageEvent{Event: "release", ID: id, VMName: vmName}) + m.recordUsage(ctx, usage.Event{Event: "release", ID: id, VMName: vmName}) return err } @@ -502,7 +503,7 @@ func (m *Manager) finalizeBatch(ctx context.Context, sbs []*types.Sandbox, ttl t } // usage lands only after the batch armed, so a rollback leaves no unterminated claim event for _, sb := range sbs { - m.recordUsage(ctx, usageEvent{ + m.recordUsage(ctx, usage.Event{ Event: "claim", ID: sb.ID, VMName: sb.VMName, KeyHash: sb.Key.Hash(), Tenant: sb.Tenant, @@ -659,7 +660,7 @@ func (m *Manager) reapOnce(ctx context.Context) { m.purgeArchiveCk(ctx, v.id, v.ck, v.tenant) m.untrack(m.pendingCks, v.ck) m.counters.reaps.Add(1) - m.recordUsage(ctx, usageEvent{Event: "reap", ID: v.id}) + m.recordUsage(ctx, usage.Event{Event: "reap", ID: v.id}) logger.Infof(ctx, "purged archived sandbox %s", v.id) case reapArchive: logSweepResult(ctx, logger, m.archive(ctx, v.sb), "archived expired sandbox "+v.id, "archive expired sandbox "+v.id) @@ -671,7 +672,7 @@ func (m *Manager) reapOnce(ctx context.Context) { m.out.Disarm(v.id, m.removeOrRetry(ctx, v.vmName, v.id, td)) m.dropSnap(ctx, v.snap) m.counters.reaps.Add(1) - m.recordUsage(ctx, usageEvent{Event: "reap", ID: v.id, VMName: v.vmName}) + m.recordUsage(ctx, usage.Event{Event: "reap", ID: v.id, VMName: v.vmName}) logger.Infof(ctx, "reaped expired sandbox %s (%s)", v.id, v.vmName) } }) @@ -704,7 +705,7 @@ func (m *Manager) purgeArchiveCk(ctx context.Context, id, ck, tenant string) { logger.Warnf(ctx, "delete archive ck %s: %v", ck, err) } m.counters.archiveDeletes.Add(1) - m.recordUsage(ctx, usageEvent{Event: "archive_delete", ID: id, Reference: ck, Tenant: tenant}) + m.recordUsage(ctx, usage.Event{Event: "archive_delete", ID: id, Reference: ck, Tenant: tenant}) } func (m *Manager) setIfLive(sb *types.Sandbox, mutate func()) (claimSnapshot, bool) { diff --git a/sandboxd/pool/egress.go b/sandboxd/pool/egress.go index c9098756f..c956b670a 100644 --- a/sandboxd/pool/egress.go +++ b/sandboxd/pool/egress.go @@ -10,6 +10,7 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/egress" "github.com/cocoonstack/sandbox/sandboxd/outbound" "github.com/cocoonstack/sandbox/sandboxd/types" + "github.com/cocoonstack/sandbox/sandboxd/usage" ) // NetRoute reports sb's route as its outbound host sees it. @@ -80,7 +81,7 @@ func (m *Manager) outboundOptions(ca *egress.CA) outbound.Options { Pooled: m.claimPooled, Record: m.recordEgress, Transfer: func(ctx context.Context, id, tenant string, ev egress.Event, sent, received int64) { - m.recordUsage(ctx, usageEvent{Event: "egress_bytes", ID: id, Tenant: tenant, Reference: ev.Host, Upstream: ev.Upstream, Sent: sent, Received: received}) + m.recordUsage(ctx, usage.Event{Event: "egress_bytes", ID: id, Tenant: tenant, Reference: ev.Host, Upstream: ev.Upstream, Sent: sent, Received: received}) }, CA: ca, LockNIC: len(m.cfg.Bridges) > 0, diff --git a/sandboxd/pool/failover.go b/sandboxd/pool/failover.go index dfcb60caa..432c26327 100644 --- a/sandboxd/pool/failover.go +++ b/sandboxd/pool/failover.go @@ -16,6 +16,7 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/store" "github.com/cocoonstack/sandbox/sandboxd/tenants" "github.com/cocoonstack/sandbox/sandboxd/types" + "github.com/cocoonstack/sandbox/sandboxd/usage" ) // AdoptArchived takes over a claim whose owner's lease expired and reports whether this node owns it now. @@ -502,7 +503,7 @@ func (m *Manager) purgeArchiveRow(ctx context.Context, rec archives.Record) { } if !rec.Consumed { m.counters.archiveDeletes.Add(1) - m.recordUsage(ctx, usageEvent{Event: "archive_delete", ID: rec.ID, Reference: rec.Checkpoint, Tenant: rec.Tenant}) + m.recordUsage(ctx, usage.Event{Event: "archive_delete", ID: rec.ID, Reference: rec.Checkpoint, Tenant: rec.Tenant}) } logger.Infof(ctx, "purged archive %s of %s", rec.ID, rec.Owner) } diff --git a/sandboxd/pool/fork.go b/sandboxd/pool/fork.go index 83c2ad8d9..1ce830533 100644 --- a/sandboxd/pool/fork.go +++ b/sandboxd/pool/fork.go @@ -8,6 +8,7 @@ import ( "time" "github.com/cocoonstack/sandbox/sandboxd/types" + "github.com/cocoonstack/sandbox/sandboxd/usage" ) // Fork clones a claimed sandbox into count children, each a fresh claim; all-or-nothing. @@ -56,7 +57,7 @@ func (m *Manager) Fork(ctx context.Context, id string, cred Cred, count int, ttl for i, c := range children { ids[i] = c.ID } - m.recordUsage(ctx, usageEvent{Event: "fork", ID: sb.ID, VMName: lockedVMName(sb), Children: ids}) + m.recordUsage(ctx, usage.Event{Event: "fork", ID: sb.ID, VMName: lockedVMName(sb), Children: ids}) return children, nil } diff --git a/sandboxd/pool/hibernate.go b/sandboxd/pool/hibernate.go index 4e464fbf1..aa6c18314 100644 --- a/sandboxd/pool/hibernate.go +++ b/sandboxd/pool/hibernate.go @@ -12,6 +12,7 @@ import ( "github.com/projecteru2/core/log" "github.com/cocoonstack/sandbox/sandboxd/types" + "github.com/cocoonstack/sandbox/sandboxd/usage" ) func (m *Manager) Hibernate(ctx context.Context, id string, cred Cred) error { @@ -207,7 +208,7 @@ func (m *Manager) wakeOrRead(ctx context.Context, sb *types.Sandbox, readerOK bo } m.counters.wakes.Add(1) m.counters.wakeNanos.Add(uint64(time.Since(wakeStart))) //nolint:gosec // durations are positive - m.recordUsage(ctx, usageEvent{Event: "wake", ID: sb.ID, VMName: sb.VMName}) + m.recordUsage(ctx, usage.Event{Event: "wake", ID: sb.ID, VMName: sb.VMName}) if proxyErr := m.out.ArmProxy(ctx, sb); proxyErr != nil { log.WithFunc("pool.wakeOrRead").Errorf(ctx, proxyErr, "arm egress proxy %s", sb.ID) } @@ -372,7 +373,7 @@ func (m *Manager) resolvePendingSnap(ctx context.Context, sb *types.Sandbox) (ad func (m *Manager) recordHibernate(ctx context.Context, sb *types.Sandbox) { m.counters.hibernates.Add(1) - m.recordUsage(ctx, usageEvent{Event: "hibernate", ID: sb.ID, VMName: sb.VMName}) + m.recordUsage(ctx, usage.Event{Event: "hibernate", ID: sb.ID, VMName: sb.VMName}) } func skipIdle(sb *types.Sandbox) bool { diff --git a/sandboxd/pool/journal.go b/sandboxd/pool/journal.go index b4ba6d692..95ab617ac 100644 --- a/sandboxd/pool/journal.go +++ b/sandboxd/pool/journal.go @@ -6,30 +6,11 @@ import ( "errors" "os" "sync" - "time" ) // journalMaxBytes rotates a journal once it crosses this size; one .1 backup is kept. const journalMaxBytes = 64 * 1024 * 1024 -// usageEvent is one line of usage.jsonl, the billing stream the platform collector folds. -type usageEvent struct { - Time time.Time `json:"t"` - Event string `json:"ev"` - ID string `json:"id"` - VMName string `json:"vm,omitempty"` - KeyHash string `json:"key,omitempty"` - Tenant string `json:"tenant,omitempty"` - Volumes []string `json:"volumes,omitempty"` - VolumesRW []string `json:"volumes_rw,omitempty"` - Children []string `json:"children,omitempty"` - Reference string `json:"ref,omitempty"` - Upstream string `json:"upstream,omitempty"` - Secret string `json:"secret,omitempty"` - Sent int64 `json:"tx,omitzero"` - Received int64 `json:"rx,omitzero"` -} - // journal is an append-only JSONL writer with size rotation; its lock only orders appends. type journal struct { mu sync.Mutex diff --git a/sandboxd/pool/liveness.go b/sandboxd/pool/liveness.go index c87398ba7..640746831 100644 --- a/sandboxd/pool/liveness.go +++ b/sandboxd/pool/liveness.go @@ -12,6 +12,7 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/engine" "github.com/cocoonstack/sandbox/sandboxd/types" + "github.com/cocoonstack/sandbox/sandboxd/usage" ) const ( @@ -149,7 +150,7 @@ func (m *Manager) recoverVMM(ctx context.Context, sb *types.Sandbox) { } if !sb.ExitRecorded { sb.ExitRecorded = true - m.recordUsage(ctx, usageEvent{Event: "vmm_exit", ID: sb.ID, VMName: sb.VMName, Tenant: sb.Tenant}) + m.recordUsage(ctx, usage.Event{Event: "vmm_exit", ID: sb.ID, VMName: sb.VMName, Tenant: sb.Tenant}) } reason := coldBootBlocker(sb, found, volumes) if reason == "" && m.vmmRestart == types.VMMRestartNone { @@ -243,7 +244,7 @@ func (m *Manager) restartLocked(ctx context.Context, sb *types.Sandbox) error { } sb.ExitRecorded = false sb.Touch() - m.recordUsage(ctx, usageEvent{Event: "vmm_restart", ID: sb.ID, VMName: sb.VMName, Tenant: sb.Tenant}) + m.recordUsage(ctx, usage.Event{Event: "vmm_restart", ID: sb.ID, VMName: sb.VMName, Tenant: sb.Tenant}) log.WithFunc("pool.restartLocked").Warnf(ctx, "restarted %s (%s) from its disk; guest memory is lost", sb.ID, sb.VMName) return nil } @@ -253,7 +254,7 @@ func (m *Manager) failLocked(ctx context.Context, sb *types.Sandbox, reason stri if live, _ := m.commitIfLive(ctx, sb, func() { sb.Failed = reason }); !live { return } - m.recordUsage(ctx, usageEvent{Event: "vmm_failed", ID: sb.ID, VMName: sb.VMName, Tenant: sb.Tenant, Reference: reason}) + m.recordUsage(ctx, usage.Event{Event: "vmm_failed", ID: sb.ID, VMName: sb.VMName, Tenant: sb.Tenant, Reference: reason}) log.WithFunc("pool.failLocked").Warnf(ctx, "sandbox %s failed: %s", sb.ID, reason) } diff --git a/sandboxd/pool/pool.go b/sandboxd/pool/pool.go index 679b7f71b..cd14662e2 100644 --- a/sandboxd/pool/pool.go +++ b/sandboxd/pool/pool.go @@ -32,6 +32,7 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/store/peer" "github.com/cocoonstack/sandbox/sandboxd/tenants" "github.com/cocoonstack/sandbox/sandboxd/types" + "github.com/cocoonstack/sandbox/sandboxd/usage" ) const ( @@ -505,11 +506,11 @@ func NewManager(ctx context.Context, cfg *config.Config, eng Engine, secrets *eg m.tplSet[rec.ID] = info } } - usage, err := newJournal(filepath.Join(cfg.DataDir, "usage.jsonl")) + usageJournal, err := newJournal(filepath.Join(cfg.DataDir, usage.JournalFile)) if err != nil { return nil, fmt.Errorf("open usage journal: %w", err) } - m.usage = usage + m.usage = usageJournal if cfg.AuditLog { audit, err := newJournal(filepath.Join(cfg.DataDir, "audit.jsonl")) if err != nil { diff --git a/sandboxd/pool/telemetry.go b/sandboxd/pool/telemetry.go index ada7fd5f4..6e948084c 100644 --- a/sandboxd/pool/telemetry.go +++ b/sandboxd/pool/telemetry.go @@ -12,6 +12,7 @@ import ( "github.com/projecteru2/core/log" "github.com/cocoonstack/sandbox/sandboxd/egress" + "github.com/cocoonstack/sandbox/sandboxd/usage" ) const ( @@ -182,11 +183,11 @@ func (m *Manager) recordEgress(ctx context.Context, id, tenant string, ev egress Tenant: tenant, Src: ev.Local, Dst: ev.Remote, }}) } - m.recordUsage(ctx, usageEvent{Event: egressEvent, ID: id, Tenant: tenant, Reference: ev.Host, Upstream: ev.Upstream, Secret: ev.Injected}) + m.recordUsage(ctx, usage.Event{Event: egressEvent, ID: id, Tenant: tenant, Reference: ev.Host, Upstream: ev.Upstream, Secret: ev.Injected}) } // recordUsage appends one billing event; failures are logged, never propagated. -func (m *Manager) recordUsage(ctx context.Context, ev usageEvent) { +func (m *Manager) recordUsage(ctx context.Context, ev usage.Event) { ev.Time = time.Now() if err := m.usage.append(ev); err != nil { log.WithFunc("pool.recordUsage").Errorf(ctx, err, "append usage event %s %s", ev.Event, ev.ID) diff --git a/sandboxd/pool/telemetry_test.go b/sandboxd/pool/telemetry_test.go index e2715401c..eb6289a71 100644 --- a/sandboxd/pool/telemetry_test.go +++ b/sandboxd/pool/telemetry_test.go @@ -16,6 +16,7 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/config" "github.com/cocoonstack/sandbox/sandboxd/egress" "github.com/cocoonstack/sandbox/sandboxd/types" + "github.com/cocoonstack/sandbox/sandboxd/usage" ) func TestUsageJournalRecordsLifecycle(t *testing.T) { @@ -53,7 +54,7 @@ func TestJournalRotationKeepsOneBackup(t *testing.T) { } j.size = journalMaxBytes - if err := j.append(usageEvent{Event: "claim", ID: "sb_1"}); err != nil { + if err := j.append(usage.Event{Event: "claim", ID: "sb_1"}); err != nil { t.Fatalf("append: %v", err) } if _, err := os.Stat(path + ".1"); err != nil { @@ -137,13 +138,13 @@ func TestTenantStampedInJournals(t *testing.T) { t.Errorf("persisted claim %+v, want tenant acme", claims[sb.ID]) } - usage, err := os.ReadFile(filepath.Join(m.dataDir, "usage.jsonl")) + journal, err := os.ReadFile(filepath.Join(m.dataDir, "usage.jsonl")) if err != nil { t.Fatalf("read usage journal: %v", err) } found := false - for line := range strings.SplitSeq(strings.TrimSpace(string(usage)), "\n") { - var ev usageEvent + for line := range strings.SplitSeq(strings.TrimSpace(string(journal)), "\n") { + var ev usage.Event if err := json.Unmarshal([]byte(line), &ev); err != nil { t.Fatalf("bad journal line %q: %v", line, err) } @@ -336,7 +337,7 @@ func TestEgressUsageEventNamesTheInjectedCredentials(t *testing.T) { if err != nil { t.Fatalf("read journal: %v", err) } - var ev usageEvent + var ev usage.Event if err = json.Unmarshal(raw, &ev); err != nil { t.Fatalf("decode %q: %v", raw, err) } @@ -422,7 +423,7 @@ func usageEventsOf(t *testing.T, m *Manager, id string) string { } var events []string for line := range strings.SplitSeq(strings.TrimSpace(string(raw)), "\n") { - var ev usageEvent + var ev usage.Event if err := json.Unmarshal([]byte(line), &ev); err != nil { t.Fatalf("bad journal line %q: %v", line, err) } diff --git a/sandboxd/pool/template.go b/sandboxd/pool/template.go index b2cc7d251..0e22d0180 100644 --- a/sandboxd/pool/template.go +++ b/sandboxd/pool/template.go @@ -15,6 +15,7 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/store" "github.com/cocoonstack/sandbox/sandboxd/types" + "github.com/cocoonstack/sandbox/sandboxd/usage" ) // TemplateInfo is the ops view of one promoted template; an empty Tenant means the operator (root). @@ -85,7 +86,7 @@ func (m *Manager) Promote(ctx context.Context, id string, cred Cred, template, t m.notifyTemplates() } m.counters.promotes.Add(1) - m.recordUsage(ctx, usageEvent{Event: "promote", ID: sb.ID, VMName: vmName, Reference: key.Template}) + m.recordUsage(ctx, usage.Event{Event: "promote", ID: sb.ID, VMName: vmName, Reference: key.Template}) return key, digest, nil } diff --git a/sandboxd/pool/volume_rw_test.go b/sandboxd/pool/volume_rw_test.go index fd3c40407..7e8715ceb 100644 --- a/sandboxd/pool/volume_rw_test.go +++ b/sandboxd/pool/volume_rw_test.go @@ -14,6 +14,7 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/config" "github.com/cocoonstack/sandbox/sandboxd/engine" "github.com/cocoonstack/sandbox/sandboxd/types" + "github.com/cocoonstack/sandbox/sandboxd/usage" ) func TestLiveWritableDetachRequiresCleanEvidence(t *testing.T) { @@ -636,7 +637,7 @@ func TestWritableDiscoveryAndUsageEvent(t *testing.T) { t.Fatalf("read usage journal: %v", err) } for line := range strings.SplitSeq(strings.TrimSpace(string(raw)), "\n") { - var event usageEvent + var event usage.Event if err := json.Unmarshal([]byte(line), &event); err != nil { t.Fatalf("decode usage event: %v", err) } diff --git a/sandboxd/pool/volume_test.go b/sandboxd/pool/volume_test.go index f79b637a3..8b6693e43 100644 --- a/sandboxd/pool/volume_test.go +++ b/sandboxd/pool/volume_test.go @@ -14,6 +14,7 @@ import ( "github.com/cocoonstack/sandbox/sandboxd/config" "github.com/cocoonstack/sandbox/sandboxd/engine" "github.com/cocoonstack/sandbox/sandboxd/types" + "github.com/cocoonstack/sandbox/sandboxd/usage" ) func TestClaimProvisionAppliesVolumesInRequestOrder(t *testing.T) { @@ -503,7 +504,7 @@ func TestVolumeClaimUsageAndScopedSummaries(t *testing.T) { t.Fatalf("read usage journal: %v", err) } for line := range strings.SplitSeq(strings.TrimSpace(string(raw)), "\n") { - var event usageEvent + var event usage.Event if err := json.Unmarshal([]byte(line), &event); err != nil { t.Fatalf("decode usage event: %v", err) } diff --git a/sandboxd/usage/usage.go b/sandboxd/usage/usage.go new file mode 100644 index 000000000..78d00b876 --- /dev/null +++ b/sandboxd/usage/usage.go @@ -0,0 +1,25 @@ +// Package usage defines the usage journal's events. +package usage + +import "time" + +// JournalFile is the usage journal's name under data_dir; it rotates to JournalFile+".1". +const JournalFile = "usage.jsonl" + +// Event is one line of usage.jsonl, the billing stream the platform collector folds. +type Event struct { + Time time.Time `json:"t"` + Event string `json:"ev"` + ID string `json:"id"` + VMName string `json:"vm,omitempty"` + KeyHash string `json:"key,omitempty"` + Tenant string `json:"tenant,omitempty"` + Volumes []string `json:"volumes,omitempty"` + VolumesRW []string `json:"volumes_rw,omitempty"` + Children []string `json:"children,omitempty"` + Reference string `json:"ref,omitempty"` + Upstream string `json:"upstream,omitempty"` + Secret string `json:"secret,omitempty"` + Sent int64 `json:"tx,omitzero"` + Received int64 `json:"rx,omitzero"` +} From 37774a73d93f7c81ed0297ec9ef6019dc6a20220 Mon Sep 17 00:00:00 2001 From: CMGS Date: Fri, 30 Oct 2026 22:50:28 +0800 Subject: [PATCH 02/16] usage: land usage events in the shared meta store and serve GET /v1/usage An outside billing service needs every node's usage events in one place. With meta_store.usage set, a background loop on each node reads usage.jsonl and its .1 backup every 5 s and lands new lines in sandboxd_usage, scoped by cell. The journal append is unchanged, so claims never wait on the database. Each node numbers its landed lines by seq; (cell, node, seq) is unique and the node's newest row (seq + line SHA-256) is its resume point, so restarts, retries, lost replies and outages converge without duplicates. Lands of a cell serialize on an advisory lock, so cursors commit in order and a reader paging by cursor never skips a row. Root reads the raw events through GET /v1/usage with tenant, since/until and cursor paging. --- .github/workflows/sandboxd.yml | 2 +- README.md | 4 +- docs/cluster.md | 7 +- docs/deploy.md | 75 ++++++++- docs/index.md | 2 +- docs/sandboxd-api.md | 33 +++- sandboxd/backends.go | 18 ++ sandboxd/config/config.go | 2 + sandboxd/config/config_test.go | 7 + sandboxd/main.go | 15 ++ sandboxd/server/server.go | 22 +++ sandboxd/server/usage.go | 71 ++++++++ sandboxd/server/usage_test.go | 103 ++++++++++++ sandboxd/usage/pg/pg.go | 129 +++++++++++++++ sandboxd/usage/pg/pg_test.go | 262 +++++++++++++++++++++++++++++ sandboxd/usage/pg/schema.sql | 25 +++ sandboxd/usage/ship.go | 238 +++++++++++++++++++++++++++ sandboxd/usage/ship_test.go | 290 +++++++++++++++++++++++++++++++++ sandboxd/usage/usage.go | 40 ++++- 19 files changed, 1333 insertions(+), 12 deletions(-) create mode 100644 sandboxd/server/usage.go create mode 100644 sandboxd/server/usage_test.go create mode 100644 sandboxd/usage/pg/pg.go create mode 100644 sandboxd/usage/pg/pg_test.go create mode 100644 sandboxd/usage/pg/schema.sql create mode 100644 sandboxd/usage/ship.go create mode 100644 sandboxd/usage/ship_test.go diff --git a/.github/workflows/sandboxd.yml b/.github/workflows/sandboxd.yml index 8a1c4ddc0..5feef8124 100644 --- a/.github/workflows/sandboxd.yml +++ b/.github/workflows/sandboxd.yml @@ -73,7 +73,7 @@ jobs: env: SANDBOXD_TEST_PG_URL: postgres://postgres:sbxci@localhost:5432/postgres?sslmode=disable run: | - GOWORK=off go test -race -count=1 -v ./tenants/... ./poolset/... | tee "$RUNNER_TEMP/pg-test.log" + GOWORK=off go test -race -count=1 -v ./tenants/... ./poolset/... ./usage/... | tee "$RUNNER_TEMP/pg-test.log" GOWORK=off go test -race -count=1 -v -run 'TestTwoNodesOnOneMetaStore|TestNodesOfACell|TestAMetaStoreNodeWithoutACell' ./pool/ | tee -a "$RUNNER_TEMP/pg-test.log" ! grep -- '--- SKIP' "$RUNNER_TEMP/pg-test.log" diff --git a/README.md b/README.md index 10eccfd47..e260e7e28 100644 --- a/README.md +++ b/README.md @@ -45,8 +45,8 @@ performance) — source in restart reconcile with VMM cold restart, memberlist mesh with redirect placement, a checkpoint store on a shared directory or S3 (sparse, content-addressed chunks), and an optional shared PostgreSQL meta store - for tenants, pools, node leases and the cross-node adoption of archived - claims and of running claims a pool replicates (`replicate_seconds`) + for tenants, pools, node leases, usage events and the cross-node adoption of + archived claims and of running claims a pool replicates (`replicate_seconds`) - `sdk/go/` — Go SDK: `Connect/New/Lookup`, `Exec/Run`, files, `Push/Pull`, sessions, `Find/Replace`, `Watch`, git verbs, `OpenPty`, `Renew/Fork/Hibernate/Promote/Checkpoint`, claim env, `DialPort/ProxyPort/PreviewURL`, diff --git a/docs/cluster.md b/docs/cluster.md index a94f4ded7..f40a47523 100644 --- a/docs/cluster.md +++ b/docs/cluster.md @@ -238,8 +238,11 @@ replication rounds write it, so the claim path is unchanged. replica retirement keeps its tombstone until the hourly sweep after that. - **Usage journal.** The journal stays per node: an adopted claim's `archive` is in the old owner's `usage.jsonl`, and its `unarchive`, `release` or - `archive_delete` is in the adopter's or the purging node's. Fold the journals - of all nodes by sandbox id. + `archive_delete` is in the adopter's or the purging node's. Fold the events + of all nodes by sandbox id. With `meta_store.usage` every node lands its + journal in the cell's [shared usage table](deploy.md#shared-usage-table), and + [`GET /v1/usage`](sandboxd-api.md#get-v1usage) on any node returns the + events of all of them. The node lease decides adoption, so a partitioned owner can lose an archived claim it still lists once its lease runs out; the epoch fences it out. A claim diff --git a/docs/deploy.md b/docs/deploy.md index aba0df01e..1ed649851 100644 --- a/docs/deploy.md +++ b/docs/deploy.md @@ -129,7 +129,7 @@ gets the intersection of the two. | `egress_usage_bytes` | `false` | append an `egress_bytes` usage event with the payload bytes of every allowed egress tunnel or request when it ends; one extra usage-journal write per connection | | `egress_ca` | unset | [HTTPS-interception](egress.md#https-interception) PKI: `root_cert` (the cluster root baked into intercepted guests; may bundle old+new roots during rotation) plus this node's `intermediate_cert`/`intermediate_key` from `sandboxd ca issue-intermediate`. Required when any pool rule sets `intercept` | | `api_token` | unset | the operator (root) credential: when set, guards the node-level endpoints (Bearer) with full access, including release-by-id cleanup. Per-sandbox tokens guard ordinary sandbox-scoped calls | -| `meta_store` | unset | `{"kind": "pg", "dsn_env": "SANDBOXD_PG_DSN"}`: keep the tenant set in one shared PostgreSQL instead of each node's `tenants.json` (see [shared tenant database](#shared-tenant-database)). `dsn_env` names the node env holding the connection string. An optional `cell` also keeps the API-applied pool set there, one per cell, instead of in each node's `pools.json` (see [shared pool set](#shared-pool-set)); without it pool targets stay per node. Needs `api_token`. Restart only | +| `meta_store` | unset | `{"kind": "pg", "dsn_env": "SANDBOXD_PG_DSN"}`: keep the tenant set in one shared PostgreSQL instead of each node's `tenants.json` (see [shared tenant database](#shared-tenant-database)). `dsn_env` names the node env holding the connection string. An optional `cell` also keeps the API-applied pool set there, one per cell, instead of in each node's `pools.json` (see [shared pool set](#shared-pool-set)); without it pool targets stay per node. An optional `usage: true` also lands this node's usage journal in one shared table and serves [`GET /v1/usage`](sandboxd-api.md#get-v1usage) from it (see [shared usage table](#shared-usage-table)). Needs `api_token`. Restart only | | `max_fork_count` | 16 | children a single `fork` may create; each is a full-RAM VM, so this bounds one request's memory blast radius to the node's capacity | | `refill_concurrency` | 0 (auto) | concurrent VM provisioning budget, shared by warm-pool refills, fork clones, and the reap/hibernate/reconcile engine batches; the batches never take the last quarter of it (at least one slot), so a mass expiry or idle sweep cannot stop refills. 0 sizes it from the node: `NumCPU*2/3` clamped to [4, 256] — a 384-core node gets 256; small nodes keep a floor of 4 | | `release_delay_seconds` | 0 | seconds a released VM waits in the removal queue before `cocoon vm rm`. The claim is dropped and journaled and its egress listener closed at release; the VM removal, the volume hold release and the egress tap unlock run on the first reap tick (5 s) after the delay, on the `refill_concurrency` budget like every other batch teardown, so a burst of releases does not compete with the claims still running. Until then the VM holds its memory and its volume reservations without counting as a claim. 0 removes inline | @@ -600,7 +600,7 @@ sandbox and checkpoint listings filter to the caller's tenant, and a tenant can delete only its own checkpoints and templates (root sees everything). Operator surfaces stay root-only — per-id sandbox reads and stats, `GET /v1/checkpoints/{id}/blob`, `GET /v1/info`, `PUT /v1/pools`, -`/v1/tenants`, `/v1/config/reload`, `POST/DELETE /v1/drain`, `/metrics`; a +`/v1/tenants`, `/v1/config/reload`, `POST/DELETE /v1/drain`, `/metrics`, `GET /v1/usage`; a tenant token there is authenticated but not authorized, so it answers 403 (a wrong token stays 401). Per-sandbox tokens are unchanged: whoever holds a sandbox's token drives that sandbox. @@ -809,6 +809,77 @@ the tenant set. - **Reverting** to config-owned pools is a `PUT` of the config pools; changing a node's `cell` takes a restart. +### Shared usage table + +With `meta_store.usage: true`, each node also lands its usage journal in one +table, and root reads the events of every node of the cell through +[`GET /v1/usage`](sandboxd-api.md#get-v1usage) on any of them. It is off by +default, so an upgrade adds no table and no writes. The table holds raw events +only; computing durations or prices stays with the consumer. + +- **Landing:** a background loop reads `usage.jsonl` and its `.1` backup every + 5 s and lands the lines it has not landed yet, up to 1000 per transaction, + until it catches up. The journal append is unchanged, so a claim, release, + hibernate or wake never waits on the database. +- **Node:** rows carry the node's mesh `node_id` (its `bind` when unset), or + its host name without a mesh. The name must be unique within the cell. +- **Schema:** sandboxd creates the table at boot when it is missing; create it + ahead with the same DDL when the node's role has no `CREATE` right: + + ```sql + CREATE TABLE IF NOT EXISTS sandboxd_usage ( + cell text NOT NULL, + cursor bigint GENERATED ALWAYS AS IDENTITY, + node text NOT NULL, + seq bigint NOT NULL CHECK (seq > 0), + line_sha256 bytea NOT NULL, + t timestamptz NOT NULL, + ev text NOT NULL, + id text NOT NULL, + vm text NOT NULL DEFAULT '', + key text NOT NULL DEFAULT '', + tenant text NOT NULL DEFAULT '', + volumes text[], + volumes_rw text[], + children text[], + ref text NOT NULL DEFAULT '', + upstream text NOT NULL DEFAULT '', + secret text NOT NULL DEFAULT '', + tx bigint NOT NULL DEFAULT 0, + rx bigint NOT NULL DEFAULT 0, + PRIMARY KEY (cell, cursor), + UNIQUE (cell, node, seq) + ); + CREATE INDEX IF NOT EXISTS sandboxd_usage_tenant ON sandboxd_usage (cell, tenant, cursor); + CREATE INDEX IF NOT EXISTS sandboxd_usage_t ON sandboxd_usage (cell, t); + ``` + +- **Columns:** the [journal line](sandboxd-api.md#usage-journal-usagejsonl)'s + keys, with an absent key as `''`, `0` or `NULL`, plus `cell` (`''` without + one), `node`, `seq` and `line_sha256`, the SHA-256 of the line. `t` keeps + microseconds, and a NUL in `ref` lands as U+FFFD. +- **Exactly once:** `seq` numbers a node's landed lines from 1 in journal + order, and `(cell, node, seq)` is unique. The node's newest row is its + resume point: at boot and after a failed land, the node reads that row's + `seq` and `line_sha256` and resumes after that line in `usage.jsonl.1` or + `usage.jsonl`. A restart, a retry or a commit whose reply was lost lands no + line twice. When neither file holds the line any more, the node lands + everything the two files hold. A line that does not decode is skipped with + a warning, and a last line without its newline waits for it. +- **Order:** a cell's lands commit one at a time under a transaction-scoped + advisory lock, so `cursor` grows in commit order and a reader that pages by + cursor never skips a row that commits later. A backfilled row gets a newer + cursor than rows with a later `t`. +- **Connection:** the table has its own query pool and no LISTEN session. +- **Outage:** claims and the journal carry on; the node logs one warning, + retries every 5 s, and backfills when the database returns. Nothing is lost + while the unlanded lines are still in `usage.jsonl` or `usage.jsonl.1`, so + an outage loses lines only after the journal rotates twice past the last + landed one (64 MiB each); the node no longer holds those either. A node + needs the database at boot, like the other shared tables. +- **Renaming** a node, or moving it to another cell, starts a new `seq` at 1: + the node lands what its journal still holds again under the new name. + ### Reloading the config `kill -HUP ` or a root `POST /v1/config/reload` diff --git a/docs/index.md b/docs/index.md index 4005a401c..d3aaec111 100644 --- a/docs/index.md +++ b/docs/index.md @@ -27,7 +27,7 @@ vsock-only I/O (hardened default); `net=egress` attaches a bridge/CNI NIC. renew, hibernate, fork, promote, checkpoints, preview, the guest-port relay, claim env, runtime volume registration, live volume changes, runtime tenants, config reload, online pool reconfigure, metrics, the usage - journal + journal and its shared table - [Replace a dataset in a running sandbox](live-volumes.md) — swapping a read-only catalog volume on a running claim with the volume API - [Hibernate with an external volume controller](volume-controller.md) — detach, diff --git a/docs/sandboxd-api.md b/docs/sandboxd-api.md index 98b166394..0e4824a4f 100644 --- a/docs/sandboxd-api.md +++ b/docs/sandboxd-api.md @@ -9,7 +9,7 @@ All bodies are JSON. Three token kinds: tenant creates is stamped with its name. Operator surfaces (the per-id sandbox reads, `PUT /v1/sandboxes/{id}/instance-metadata`, `GET /v1/info`, `PUT /v1/pools`, `PUT/DELETE /v1/volumes/{name}`, `POST/DELETE /v1/drain`, - `GET /metrics`, `GET /v1/checkpoints/{id}/blob`) + `GET /metrics`, `GET /v1/usage`, `GET /v1/checkpoints/{id}/blob`) answer a tenant token `403` — authenticated but not authorized; an unknown token stays `401`. On a node with a shared [`meta_store`](deploy.md#shared-tenant-database), a tenant token the node @@ -1035,7 +1035,36 @@ point, and `vmm_failed` is informational; hibernated storage seconds = VM's RAM. An interval left open by a crash clamps to the claim's deadline; reconcile records nothing for the claims it drops (those whose VM record is gone), so a restart leaves those intervals without a terminal event. The `vm` name joins -cocoon's machine-level metering ledger for audit cross-checks. +cocoon's machine-level metering ledger for audit cross-checks. With +`meta_store.usage`, the node also lands every line in the cell's +[shared usage table](deploy.md#shared-usage-table), read through +[`GET /v1/usage`](#get-v1usage). + +## GET /v1/usage + +Auth: root only (tenant tokens get 403). Reads the raw usage events of every +node of the cell from the [shared usage table](deploy.md#shared-usage-table). + +- `GET /v1/usage?tenant=&since=