refactor: localize qmp collector metrics

This commit is contained in:
joshsiv-mulga
2026-10-01 14:06:10 +10:00
parent b975f1ba9d
commit 08d9507701
5 changed files with 26 additions and 15 deletions
+9
View File
@@ -192,6 +192,7 @@ while the branch is in flight; they do not describe the umbrella repository's
| `f4cb550c9` | `spinifex/types/ebs.go` (GP3 policy constants) | `spinifex/domains/ec2/ebs/policy/gp3.go` | The supported GP3 product envelope is EC2 EBS domain policy, not generic runtime or Viperblockd wire state; `gp3_test.go` pins the deployed values. |
| `adb55de33` | `spinifex/types/ebs.go` (`NBDTransport`) | `spinifex/services/viperblockd/nbd_transport.go` | NBD endpoint transport selection is Viperblockd-local delivery configuration, not generic EBS state or a provider contract; `nbd_transport_test.go` pins the socket and TCP forms. |
| `94a6e7af2` | `spinifex/types/ebs.go` (`EBSConfigUpdateRequest`, `EBSConfigUpdateResponse`) | `contracts/viperblockd/legacy/v1/config_update.go` | Legacy configuration updates have deployed queue and volume-addressed NATS routes; the versioned compatibility contract is expressly transitional and separate from `providers/ebs`. |
| `b975f1ba9` | Remaining `spinifex/types/ebs.go` payloads | `runtime/compute/vm/ebs_requests.go` and `contracts/viperblockd/legacy/v1/volume.go` | The mutex-bearing attachment collection is VM-owned; deployed mount, unmount and delete routes plus their JSON payloads are versioned Viperblockd legacy compatibility. Contract tests pin every route and JSON shape. |
The EC2 contract row is intentionally different from the directory moves: it creates
a compatibility boundary. `ec2.cmd.*` retains its deployed one-token NATS
@@ -199,6 +200,14 @@ subscription and permission shape, while callers use
`InstanceCommandSubject` rather than reconstructing the subject. The
contract's README states the route, scope and change rule.
## Pending paths discovered after the baseline
`spinifex/cloud/exoscale` was introduced after the initial inventory. It is an
external-cloud adapter and has the same target classification as OCI:
`spinifex/providers/cloud/exoscale`. It remains at its current path until a
dedicated structural move inventories and rewrites its callers; it is not
folded into the OCI relocation retrospectively.
## Recording rule
For every later slice, add the old and new path, source commit, focused
@@ -7,7 +7,6 @@ import (
"sync"
"time"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
"go.opentelemetry.io/otel"
@@ -62,7 +61,7 @@ func startBridge(nc *nats.Conn) (func(), error) {
}
func (b *bridge) handle(msg *nats.Msg) {
var batch types.TelemetryBatch
var batch TelemetryBatch
if err := json.Unmarshal(msg.Data, &batch); err != nil {
slog.Warn("qmp-collector bridge: bad batch", "subject", msg.Subject, "err", err)
return
@@ -4,7 +4,6 @@ import (
"encoding/json"
"testing"
"github.com/mulgadc/spinifex/spinifex/types"
"github.com/mulgadc/spinifex/spinifex/utils"
"github.com/nats-io/nats.go"
"github.com/stretchr/testify/assert"
@@ -29,14 +28,14 @@ func newTestBridge() *bridge {
func handleOne(t *testing.T, labels map[string]string) attribute.Set {
t.Helper()
b := newTestBridge()
batch := types.TelemetryBatch{
batch := TelemetryBatch{
PeriodSeconds: 60,
Node: "node-1",
Series: []types.TelemetrySeries{{Name: "goanna_ec2_cpu_utilization", Labels: labels, Value: 12}},
Series: []TelemetrySeries{{Name: "goanna_ec2_cpu_utilization", Labels: labels, Value: 12}},
}
data, err := json.Marshal(batch)
require.NoError(t, err)
b.handle(&nats.Msg{Subject: types.MetricsEC2SubjectPrefix + "i-test", Data: data})
b.handle(&nats.Msg{Subject: MetricsEC2SubjectPrefix + "i-test", Data: data})
entries := b.series["goanna_ec2_cpu_utilization"]
require.Len(t, entries, 1, "one series in, one series stored")
@@ -1,8 +1,13 @@
package types
package qmpcollector
// 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.
//
// The stream is not a Spinifex-owned versioned telemetry contract yet: its
// eventual consumer lives in Goanna. Keep compatibility with the deployed
// subject and JSON shape, but make a cross-repository contract decision before
// treating this local representation as a public API.
const MetricsEC2SubjectPrefix = "metrics.ec2."
// TelemetrySeries is one CloudWatch-mappable datapoint. Names and labels
@@ -14,7 +14,6 @@ import (
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"
"github.com/nats-io/nats.go"
"go.opentelemetry.io/otel"
@@ -187,7 +186,7 @@ func (p *poller) tick(ctx context.Context) {
utils.MarkSpanError(span, err)
return
}
if err := p.nc.Publish(types.MetricsEC2SubjectPrefix+meta.InstanceID, data); err != nil {
if err := p.nc.Publish(MetricsEC2SubjectPrefix+meta.InstanceID, data); err != nil {
slog.WarnContext(ctx, "qmp-collector: publish failed", "instanceId", meta.InstanceID, "err", err)
utils.MarkSpanError(span, err)
return
@@ -299,16 +298,16 @@ 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 telemetryv1.GuestTelemetryMeta, node string, prev, cur *sample) (types.TelemetryBatch, bool) {
func buildBatch(meta telemetryv1.GuestTelemetryMeta, node string, prev, cur *sample) (TelemetryBatch, bool) {
elapsed := cur.at.Sub(prev.at).Seconds()
if elapsed <= 0 {
return types.TelemetryBatch{}, false
return TelemetryBatch{}, false
}
if cur.cpuJiffies < prev.cpuJiffies ||
cur.rdBytes < prev.rdBytes || cur.wrBytes < prev.wrBytes ||
cur.rdOps < prev.rdOps || cur.wrOps < prev.wrOps ||
cur.rxBytes < prev.rxBytes || cur.txBytes < prev.txBytes {
return types.TelemetryBatch{}, false
return TelemetryBatch{}, false
}
labels := map[string]string{
@@ -325,7 +324,7 @@ func buildBatch(meta telemetryv1.GuestTelemetryMeta, node string, prev, cur *sam
cpuPct = 100
}
series := []types.TelemetrySeries{
series := []TelemetrySeries{
{Name: "goanna_ec2_cpu_utilization", Labels: labels, Value: cpuPct, Unit: "Percent"},
{Name: "goanna_ec2_network_in_bytes", Labels: labels, Value: float64(cur.txBytes - prev.txBytes), Unit: "Bytes"},
{Name: "goanna_ec2_network_out_bytes", Labels: labels, Value: float64(cur.rxBytes - prev.rxBytes), Unit: "Bytes"},
@@ -335,13 +334,13 @@ func buildBatch(meta telemetryv1.GuestTelemetryMeta, node string, prev, cur *sam
{Name: "goanna_ec2_disk_write_ops", Labels: labels, Value: float64(cur.wrOps - prev.wrOps), Unit: "Count"},
}
if cur.balloonOK {
series = append(series, types.TelemetrySeries{
series = append(series, TelemetrySeries{
Name: "goanna_ec2_memory_actual_bytes", Labels: labels,
Value: float64(cur.balloon), Unit: "Bytes",
})
}
return types.TelemetryBatch{
return TelemetryBatch{
TS: cur.at.Unix(),
PeriodSeconds: int(elapsed + 0.5),
Node: node,