refactor: establish telemetry sidecar contract

This commit is contained in:
joshsiv-mulga
2026-09-30 16:05:35 +10:00
parent 5d88a33998
commit 78fd017e03
13 changed files with 93 additions and 48 deletions
+12
View File
@@ -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-<instance-id>.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.
+19
View File
@@ -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-<instance-id>.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"`
}
+28
View File
@@ -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))
}
+1 -2
View File
@@ -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 <name>` (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.
+1
View File
@@ -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
@@ -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,
@@ -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)
}
@@ -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
@@ -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)
@@ -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
@@ -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")
}
@@ -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.<instance-id>. A bridge goroutine forwards the series to
-16
View File
@@ -1,21 +1,5 @@
package types
// GuestTelemetryMeta is the qmp-collector's per-VM discovery record: written
// as qmp-telemetry-<instance-id>.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.<instance-id>. Goanna is the eventual consumer; the
// collector's OTLP bridge taps it for the operator plane meanwhile.