From 78fd017e03d1d729d52fd28e7664d90bc42a919e Mon Sep 17 00:00:00 2001 From: joshsiv-mulga Date: Wed, 30 Sep 2026 16:05:35 +1000 Subject: [PATCH] refactor: establish telemetry sidecar contract --- contracts/telemetry/v1/README.md | 12 ++++++++ contracts/telemetry/v1/guest_meta.go | 19 +++++++++++++ contracts/telemetry/v1/guest_meta_test.go | 28 +++++++++++++++++++ docs/COMMANDS.md | 3 +- docs/PACKAGE_BOUNDARY_MIGRATION.md | 1 + spinifex/runtime/compute/vm/telemetry_meta.go | 4 +-- .../runtime/compute/vm/telemetry_meta_test.go | 6 ++-- .../runtime/roles/qmpcollector/collector.go | 6 ++-- .../roles/qmpcollector/collector_test.go | 10 +++---- spinifex/runtime/roles/qmpcollector/poller.go | 13 +++++---- .../runtime/roles/qmpcollector/poller_test.go | 20 ++++++------- .../roles/qmpcollector/qmpcollector.go | 3 +- spinifex/types/telemetry.go | 16 ----------- 13 files changed, 93 insertions(+), 48 deletions(-) create mode 100644 contracts/telemetry/v1/README.md create mode 100644 contracts/telemetry/v1/guest_meta.go create mode 100644 contracts/telemetry/v1/guest_meta_test.go diff --git a/contracts/telemetry/v1/README.md b/contracts/telemetry/v1/README.md new file mode 100644 index 000000000..e535b1fb4 --- /dev/null +++ b/contracts/telemetry/v1/README.md @@ -0,0 +1,12 @@ +# Telemetry v1 contracts + +`GuestTelemetryMeta` is the JSON sidecar contract between the VM runtime and +the qmp-collector role. The writer creates and refreshes +`qmp-telemetry-.json`; the collector reads it to locate the +dedicated QMP socket, determine the monitoring interval, and select VPC data +plane taps. Do not change this JSON shape incompatibly; add a new version when +required. + +The `metrics.ec2.*` NATS batch schema is deliberately not owned here yet. +Goanna consumes that stream in another repository, so its ownership requires a +cross-repository contract decision rather than a Spinifex-only move. diff --git a/contracts/telemetry/v1/guest_meta.go b/contracts/telemetry/v1/guest_meta.go new file mode 100644 index 000000000..72437d819 --- /dev/null +++ b/contracts/telemetry/v1/guest_meta.go @@ -0,0 +1,19 @@ +// Package telemetryv1 defines version 1 of cross-process telemetry contracts. +package telemetryv1 + +// GuestTelemetryMeta is the qmp-collector's per-VM discovery record. The VM +// runtime writes qmp-telemetry-.json next to the dedicated +// telemetry QMP socket; qmp-collector reads and refreshes it after VM or ENI +// lifecycle changes. +type GuestTelemetryMeta struct { + InstanceID string `json:"instance_id"` + AccountID string `json:"account_id,omitempty"` + VCPUs int `json:"vcpus"` + // PeriodSeconds is the collection interval: 60 for detailed monitoring, + // 300 for basic — the EC2 monitoring tiers. + PeriodSeconds int `json:"period_seconds"` + // Taps are the VPC data-plane tap devices whose host-side counters yield + // NetworkIn/Out. Control-plane NICs (mgmt, dev hostfwd) are excluded. + Taps []string `json:"taps,omitempty"` + Socket string `json:"socket"` +} diff --git a/contracts/telemetry/v1/guest_meta_test.go b/contracts/telemetry/v1/guest_meta_test.go new file mode 100644 index 000000000..c1d87a6e1 --- /dev/null +++ b/contracts/telemetry/v1/guest_meta_test.go @@ -0,0 +1,28 @@ +package telemetryv1 + +import ( + "encoding/json" + "testing" + + "github.com/stretchr/testify/require" +) + +func TestGuestTelemetryMetaJSONContract(t *testing.T) { + data, err := json.Marshal(GuestTelemetryMeta{ + InstanceID: "i-0123456789abcdef0", + AccountID: "123456789012", + VCPUs: 2, + PeriodSeconds: 60, + Taps: []string{"tap0abc", "tap0def"}, + Socket: "/run/spinifex/qmp-telemetry-i-0123456789abcdef0.sock", + }) + require.NoError(t, err) + require.JSONEq(t, `{ + "instance_id":"i-0123456789abcdef0", + "account_id":"123456789012", + "vcpus":2, + "period_seconds":60, + "taps":["tap0abc","tap0def"], + "socket":"/run/spinifex/qmp-telemetry-i-0123456789abcdef0.sock" + }`, string(data)) +} diff --git a/docs/COMMANDS.md b/docs/COMMANDS.md index f16e61d42..d85ab41f5 100644 --- a/docs/COMMANDS.md +++ b/docs/COMMANDS.md @@ -23,7 +23,7 @@ Every service is `PartOf=spinifex.target` and ordered by its own `After=`/`Wants | `spinifex-awsgw.service` | `spx service awsgw start` | AWS-compatible gateway: SigV4 auth, IAM policy enforcement, TLS. Also serves the Private Admin API | | `spinifex-vpcd.service` | `spx service vpcd start` | VPC daemon: translates `vpc.*` NATS events to OVN logical switches, ports and routers | | `spinifex-ui.service` | `spx service spinifex-ui start` | Web UI serving the React frontend | -| `spinifex-qmp-collector.service` | `spx service qmp-collector start` | Polls per-VM telemetry QMP sockets and tap counters, publishes to NATS `metrics.ec2.*` | +| `spinifex-qmp-collector.service` | `spx service qmp-collector start` | Reads the `contracts/telemetry/v1` per-VM sidecar, polls telemetry QMP sockets and tap counters, publishes to NATS `metrics.ec2.*` | | `spinifex-shutdown.service` | `/bin/true`; the work is its `ExecStop` | Runs `spx admin node drain --local --unless-restarting`. `After=` every service, so it stops first and drains guests while the stack is still up | | `spinifex-nats-watchdog.service` / `.timer` | `nats-js-watchdog.sh` | JetStream liveness probe every 30s. `try-restart`s NATS when JetStream stops accepting writes while the process stays alive — what `Restart=on-failure` cannot catch | | `spinifex-firewall.service` | `spinifex-firewall-apply` | Oneshot nftables input policy, `WantedBy=sysinit.target` rather than the target. No `ExecStop`: remove it with `nft delete table inet spinifex_filter` | @@ -210,4 +210,3 @@ Errors are `{"error":{"code":…,"message":…},"requestId":…}`. `OperationInP Each method is a separate grant, and `spx admin principal create` decides which a credential holds. `spx admin principal create operator` grants all six — the operator and load-test credential. `spx admin principal create signup --grant CreateAccount` grants one, which is what an internet-facing signup form should hold: that key cannot reach deletion, so a compromise of the form's host cannot remove a tenant. Separate principals are the point rather than an accident: revoking either costs nothing else. The surface is unreachable until an operator runs `spx admin principal create`; there is no config toggle and no default credential for it. The super-admin account's own `admin` user does hold these grants already, through the `AdministratorAccess` policy every account's administrator carries (`Action: "*"`), so its bootstrap key can call the surface without any further setup — but it is unscoped and unrotatable without locking out the operator, which is why automation should use a named principal. Revocation is `spx admin principal revoke ` (or `aws iam delete-access-key`), which is immediate and cluster-wide. `[signup] max_accounts` in `awsgw.toml` caps how many accounts may exist; it defaults to 128 when the section is absent, and 0 means uncapped. - diff --git a/docs/PACKAGE_BOUNDARY_MIGRATION.md b/docs/PACKAGE_BOUNDARY_MIGRATION.md index f4cd38b4a..7918650ed 100644 --- a/docs/PACKAGE_BOUNDARY_MIGRATION.md +++ b/docs/PACKAGE_BOUNDARY_MIGRATION.md @@ -185,6 +185,7 @@ while the branch is in flight; they do not describe the umbrella repository's | `aa0c36ff4` | `spinifex/types/ec2.go` | `contracts/ec2/v1` | The EC2 instance-command subject and JSON payload are an explicit versioned cross-process contract; `commands_test.go` is its compatibility evidence. | | `c239fd897` | `spinifex/types/eni.go` | `spinifex/runtime/compute/vm/eni_requests.go` | PCIe hot-plug slot allocation is mutex-bearing, per-VM runtime state rather than a shared type or wire contract; `record_test.go` preserves its restart semantics. | | `6d6275e78` | `spinifex/types/cluster.go` (`NodeHealthResponse`) | `contracts/cluster/v1/health.go` | Node health is a versioned cluster control-plane representation shared by the per-node NATS route and daemon HTTPS `/health`; `health_test.go` pins both wire forms. | +| `53a614717` | `spinifex/types/cluster.go` (`SharedClusterData`) | `spinifex/daemon/config_hash.go` | Shared-config hash input is daemon-local implementation data; focused config-hash tests preserve deterministic, shared-only semantics. | The EC2 contract row is intentionally different from the directory moves: it creates a compatibility boundary. `ec2.cmd.*` retains its deployed one-token NATS diff --git a/spinifex/runtime/compute/vm/telemetry_meta.go b/spinifex/runtime/compute/vm/telemetry_meta.go index cf0ec64fc..5396de73d 100644 --- a/spinifex/runtime/compute/vm/telemetry_meta.go +++ b/spinifex/runtime/compute/vm/telemetry_meta.go @@ -7,7 +7,7 @@ import ( "sort" "github.com/aws/aws-sdk-go/aws" - "github.com/mulgadc/spinifex/spinifex/types" + telemetryv1 "github.com/mulgadc/spinifex/contracts/telemetry/v1" "github.com/mulgadc/spinifex/spinifex/utils" ) @@ -57,7 +57,7 @@ func writeTelemetryMeta(v *VM) error { if v.Config.TelemetryQMPSocket == "" { return nil } - meta := types.GuestTelemetryMeta{ + meta := telemetryv1.GuestTelemetryMeta{ InstanceID: v.ID, AccountID: v.AccountID, VCPUs: v.Config.CPUCount, diff --git a/spinifex/runtime/compute/vm/telemetry_meta_test.go b/spinifex/runtime/compute/vm/telemetry_meta_test.go index e718815e5..bb7b8300c 100644 --- a/spinifex/runtime/compute/vm/telemetry_meta_test.go +++ b/spinifex/runtime/compute/vm/telemetry_meta_test.go @@ -8,7 +8,7 @@ import ( "github.com/aws/aws-sdk-go/aws" "github.com/aws/aws-sdk-go/service/ec2" - "github.com/mulgadc/spinifex/spinifex/types" + telemetryv1 "github.com/mulgadc/spinifex/contracts/telemetry/v1" "github.com/mulgadc/spinifex/spinifex/utils" ) @@ -75,7 +75,7 @@ func TestTelemetryMetaRoundtrip(t *testing.T) { if err != nil { t.Fatalf("read back: %v", err) } - var meta types.GuestTelemetryMeta + var meta telemetryv1.GuestTelemetryMeta if err := json.Unmarshal(data, &meta); err != nil { t.Fatalf("unmarshal: %v", err) } @@ -105,7 +105,7 @@ func TestRefreshTelemetryMetaCarriesTierChange(t *testing.T) { if err != nil { t.Fatalf("read back: %v", err) } - var meta types.GuestTelemetryMeta + var meta telemetryv1.GuestTelemetryMeta if err := json.Unmarshal(data, &meta); err != nil { t.Fatalf("unmarshal: %v", err) } diff --git a/spinifex/runtime/roles/qmpcollector/collector.go b/spinifex/runtime/roles/qmpcollector/collector.go index 159bfa39e..653165902 100644 --- a/spinifex/runtime/roles/qmpcollector/collector.go +++ b/spinifex/runtime/roles/qmpcollector/collector.go @@ -14,7 +14,7 @@ import ( "strings" "time" - "github.com/mulgadc/spinifex/spinifex/types" + telemetryv1 "github.com/mulgadc/spinifex/contracts/telemetry/v1" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" ) @@ -93,8 +93,8 @@ func (c *collector) reconcile(ctx context.Context) { } } -func readMeta(path string) (types.GuestTelemetryMeta, error) { - var meta types.GuestTelemetryMeta +func readMeta(path string) (telemetryv1.GuestTelemetryMeta, error) { + var meta telemetryv1.GuestTelemetryMeta data, err := os.ReadFile(path) if err != nil { return meta, err diff --git a/spinifex/runtime/roles/qmpcollector/collector_test.go b/spinifex/runtime/roles/qmpcollector/collector_test.go index 59d64c333..260601507 100644 --- a/spinifex/runtime/roles/qmpcollector/collector_test.go +++ b/spinifex/runtime/roles/qmpcollector/collector_test.go @@ -7,11 +7,11 @@ import ( "testing" "time" - "github.com/mulgadc/spinifex/spinifex/types" + telemetryv1 "github.com/mulgadc/spinifex/contracts/telemetry/v1" "github.com/mulgadc/spinifex/spinifex/utils" ) -func writeMeta(t *testing.T, dir string, meta types.GuestTelemetryMeta) string { +func writeMeta(t *testing.T, dir string, meta telemetryv1.GuestTelemetryMeta) string { t.Helper() data, err := json.Marshal(meta) if err != nil { @@ -41,11 +41,11 @@ func TestReconcile(t *testing.T) { // Live VM: metadata + socket present. liveSock := filepath.Join(dir, utils.QMPTelemetryPrefix+"i-live.sock") touch(t, liveSock) - writeMeta(t, dir, types.GuestTelemetryMeta{ + writeMeta(t, dir, telemetryv1.GuestTelemetryMeta{ InstanceID: "i-live", Socket: liveSock, PeriodSeconds: 300}) // Stale VM: metadata but no socket (unclean QEMU exit) — must be GC'd. - stalePath := writeMeta(t, dir, types.GuestTelemetryMeta{ + stalePath := writeMeta(t, dir, telemetryv1.GuestTelemetryMeta{ InstanceID: "i-stale", Socket: filepath.Join(dir, "gone.sock"), PeriodSeconds: 300}) // Garbage file matching the glob — skipped, not fatal. @@ -70,7 +70,7 @@ func TestReconcile(t *testing.T) { } // Metadata refresh (ENI hotplug) must update the running poller in place. - writeMeta(t, dir, types.GuestTelemetryMeta{ + writeMeta(t, dir, telemetryv1.GuestTelemetryMeta{ InstanceID: "i-live", Socket: liveSock, PeriodSeconds: 300, Taps: []string{"tapnew"}}) c.reconcile(ctx) diff --git a/spinifex/runtime/roles/qmpcollector/poller.go b/spinifex/runtime/roles/qmpcollector/poller.go index c68ca7f53..8c88995c9 100644 --- a/spinifex/runtime/roles/qmpcollector/poller.go +++ b/spinifex/runtime/roles/qmpcollector/poller.go @@ -12,6 +12,7 @@ import ( "sync" "time" + telemetryv1 "github.com/mulgadc/spinifex/contracts/telemetry/v1" "github.com/mulgadc/spinifex/spinifex/runtime/compute/qmp" "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/utils" @@ -56,17 +57,17 @@ type poller struct { retier chan struct{} mu sync.Mutex - meta types.GuestTelemetryMeta + meta telemetryv1.GuestTelemetryMeta prev *sample } -func newPoller(cfg *Config, nc *nats.Conn, meta types.GuestTelemetryMeta) *poller { +func newPoller(cfg *Config, nc *nats.Conn, meta telemetryv1.GuestTelemetryMeta) *poller { return &poller{cfg: cfg, nc: nc, meta: meta, retier: make(chan struct{}, 1)} } // updateMeta refreshes taps/period after an ENI hot-plug or monitoring-tier // rewrite, waking the run loop when the period moved. -func (p *poller) updateMeta(meta types.GuestTelemetryMeta) { +func (p *poller) updateMeta(meta telemetryv1.GuestTelemetryMeta) { p.mu.Lock() retier := meta.PeriodSeconds != p.meta.PeriodSeconds p.meta = meta @@ -82,7 +83,7 @@ func (p *poller) updateMeta(meta types.GuestTelemetryMeta) { } } -func (p *poller) snapshotMeta() types.GuestTelemetryMeta { +func (p *poller) snapshotMeta() telemetryv1.GuestTelemetryMeta { p.mu.Lock() defer p.mu.Unlock() return p.meta @@ -196,7 +197,7 @@ func (p *poller) tick(ctx context.Context) { // collect dials the telemetry socket fresh each tick (no held connection to // leak across QEMU restarts) and reads QMP + procfs + sysfs counters. -func (p *poller) collect(meta types.GuestTelemetryMeta) (*sample, error) { +func (p *poller) collect(meta telemetryv1.GuestTelemetryMeta) (*sample, error) { client, err := qmp.NewQMPClient(meta.Socket) if err != nil { return nil, fmt.Errorf("dial telemetry socket: %w", err) @@ -298,7 +299,7 @@ func readSysfsCounter(sysRoot, iface, counter string) (uint64, error) { // buildBatch converts two snapshots into the locked goanna_ec2_* series set. // ok is false when a counter regressed (QEMU restart between ticks) — the // fresh snapshot then serves as the new baseline and nothing is published. -func buildBatch(meta types.GuestTelemetryMeta, node string, prev, cur *sample) (types.TelemetryBatch, bool) { +func buildBatch(meta telemetryv1.GuestTelemetryMeta, node string, prev, cur *sample) (types.TelemetryBatch, bool) { elapsed := cur.at.Sub(prev.at).Seconds() if elapsed <= 0 { return types.TelemetryBatch{}, false diff --git a/spinifex/runtime/roles/qmpcollector/poller_test.go b/spinifex/runtime/roles/qmpcollector/poller_test.go index 11952a662..e6defe9d5 100644 --- a/spinifex/runtime/roles/qmpcollector/poller_test.go +++ b/spinifex/runtime/roles/qmpcollector/poller_test.go @@ -7,7 +7,7 @@ import ( "testing" "time" - "github.com/mulgadc/spinifex/spinifex/types" + telemetryv1 "github.com/mulgadc/spinifex/contracts/telemetry/v1" ) func writeFile(t *testing.T, path, content string) { @@ -93,7 +93,7 @@ func TestReadSysfsCounter(t *testing.T) { } func TestBuildBatch(t *testing.T) { - meta := types.GuestTelemetryMeta{ + meta := telemetryv1.GuestTelemetryMeta{ InstanceID: "i-0abc", AccountID: "123456789012", VCPUs: 2, PeriodSeconds: 60, } base := time.Unix(1000, 0) @@ -147,7 +147,7 @@ func TestBuildBatch(t *testing.T) { } func TestBuildBatchCounterRegression(t *testing.T) { - meta := types.GuestTelemetryMeta{InstanceID: "i-0abc", VCPUs: 1, PeriodSeconds: 60} + meta := telemetryv1.GuestTelemetryMeta{InstanceID: "i-0abc", VCPUs: 1, PeriodSeconds: 60} base := time.Unix(1000, 0) prev := &sample{at: base, cpuJiffies: 5000, rdBytes: 900} // QEMU restarted: counters reset below prev. @@ -159,7 +159,7 @@ func TestBuildBatchCounterRegression(t *testing.T) { } func TestBuildBatchNoBalloon(t *testing.T) { - meta := types.GuestTelemetryMeta{InstanceID: "i-0abc", VCPUs: 1, PeriodSeconds: 60} + meta := telemetryv1.GuestTelemetryMeta{InstanceID: "i-0abc", VCPUs: 1, PeriodSeconds: 60} base := time.Unix(1000, 0) batch, ok := buildBatch(meta, "", &sample{at: base}, &sample{at: base.Add(time.Minute)}) @@ -187,7 +187,7 @@ func TestPollerPeriod(t *testing.T) { {17, 300 * time.Second}, // garbage -> basic } for _, tt := range tests { - p := newPoller(&Config{}, nil, types.GuestTelemetryMeta{PeriodSeconds: tt.secs}) + p := newPoller(&Config{}, nil, telemetryv1.GuestTelemetryMeta{PeriodSeconds: tt.secs}) if got := p.period(); got != tt.want { t.Errorf("period(%d) = %v, want %v", tt.secs, got, tt.want) } @@ -198,7 +198,7 @@ func TestPollerPeriod(t *testing.T) { // out the interval being replaced, so moving a 300s instance to the detailed // tier would take five minutes to take effect. func TestUpdateMetaSignalsOnlyTierChanges(t *testing.T) { - p := newPoller(&Config{}, nil, types.GuestTelemetryMeta{InstanceID: "i-0aaa", PeriodSeconds: 300}) + p := newPoller(&Config{}, nil, telemetryv1.GuestTelemetryMeta{InstanceID: "i-0aaa", PeriodSeconds: 300}) signalled := func() bool { select { @@ -209,12 +209,12 @@ func TestUpdateMetaSignalsOnlyTierChanges(t *testing.T) { } } - p.updateMeta(types.GuestTelemetryMeta{InstanceID: "i-0aaa", PeriodSeconds: 300, Taps: []string{"tap0"}}) + p.updateMeta(telemetryv1.GuestTelemetryMeta{InstanceID: "i-0aaa", PeriodSeconds: 300, Taps: []string{"tap0"}}) if signalled() { t.Error("a taps-only rewrite must not signal a tier change") } - p.updateMeta(types.GuestTelemetryMeta{InstanceID: "i-0aaa", PeriodSeconds: 60}) + p.updateMeta(telemetryv1.GuestTelemetryMeta{InstanceID: "i-0aaa", PeriodSeconds: 60}) if !signalled() { t.Fatal("moving to the detailed tier must signal") } @@ -224,8 +224,8 @@ func TestUpdateMetaSignalsOnlyTierChanges(t *testing.T) { // The signal is coalesced, so a poller that has not woken yet cannot // accumulate a backlog of resets. - p.updateMeta(types.GuestTelemetryMeta{InstanceID: "i-0aaa", PeriodSeconds: 300}) - p.updateMeta(types.GuestTelemetryMeta{InstanceID: "i-0aaa", PeriodSeconds: 60}) + p.updateMeta(telemetryv1.GuestTelemetryMeta{InstanceID: "i-0aaa", PeriodSeconds: 300}) + p.updateMeta(telemetryv1.GuestTelemetryMeta{InstanceID: "i-0aaa", PeriodSeconds: 60}) if !signalled() { t.Fatal("a pending tier change must be visible") } diff --git a/spinifex/runtime/roles/qmpcollector/qmpcollector.go b/spinifex/runtime/roles/qmpcollector/qmpcollector.go index dc378861b..44b2c88d4 100644 --- a/spinifex/runtime/roles/qmpcollector/qmpcollector.go +++ b/spinifex/runtime/roles/qmpcollector/qmpcollector.go @@ -1,5 +1,6 @@ // Package qmpcollector is the per-node guest-metrics producer: -// it discovers running VMs via their qmp-telemetry-*.json sidecar files, polls +// it discovers running VMs via the telemetryv1 qmp-telemetry-*.json sidecar +// contract, polls // each VM's dedicated telemetry QMP socket (never the manager's control // socket) plus host tap counters, and publishes CloudWatch-mappable series to // NATS metrics.ec2.. A bridge goroutine forwards the series to diff --git a/spinifex/types/telemetry.go b/spinifex/types/telemetry.go index e0baa6045..165867c2d 100644 --- a/spinifex/types/telemetry.go +++ b/spinifex/types/telemetry.go @@ -1,21 +1,5 @@ package types -// GuestTelemetryMeta is the qmp-collector's per-VM discovery record: written -// as qmp-telemetry-.json in the runtime dir next to the VM's -// telemetry QMP socket, rewritten on ENI hot-plug, removed on stop/terminate. -type GuestTelemetryMeta struct { - InstanceID string `json:"instance_id"` - AccountID string `json:"account_id,omitempty"` - VCPUs int `json:"vcpus"` - // PeriodSeconds is the collection interval: 60 for detailed monitoring, - // 300 for basic — the EC2 monitoring tiers. - PeriodSeconds int `json:"period_seconds"` - // Taps are the VPC data-plane tap devices whose host-side counters - // yield NetworkIn/Out. Control-plane NICs (mgmt, dev hostfwd) excluded. - Taps []string `json:"taps,omitempty"` - Socket string `json:"socket"` -} - // MetricsEC2SubjectPrefix is the NATS subject family for per-instance guest // metrics: metrics.ec2.. Goanna is the eventual consumer; the // collector's OTLP bridge taps it for the operator plane meanwhile.