feat(obs): emit OTLP spans from qmp-collector and spinifex-ui (#524)

Both services already run otelsetup.Init and ship logs/metrics but never
opened a span, so they were invisible in the Kibana APM service map.
qmp-collector now opens a root span per per-instance poll tick (marks
error on QMP collect/publish failure); spinifex-ui now wraps its HTTP
handler chain with otelsetup.HTTPMiddleware and its reverse-proxy
transport with otelhttp, so UI requests get a server span and proxied
calls to awsgw/predastore carry traceparent instead of stopping the
trace at the UI hop.

nats and eks are intentionally left uninstrumented in this pass: nats is
the vendored message-broker process with no message-level hook to attach
spans to (the traffic it carries is already traced end-to-end at its
producer and consumer endpoints); eks request/response calls already
produce spans today under service.name=spinifex-daemon since the
handlers run in that process, and its async reconciler/state-machine
goroutines need a separate span design across a much larger surface.

spinifex/services/qmpcollector/poller.go

@@ -14,13 +14,20 @@ import (
 
 	"github.com/mulgadc/spinifex/spinifex/qmp"
 	"github.com/mulgadc/spinifex/spinifex/types"
+	"github.com/mulgadc/spinifex/spinifex/utils"
 	"github.com/nats-io/nats.go"
+	"go.opentelemetry.io/otel"
+	"go.opentelemetry.io/otel/attribute"
+	"go.opentelemetry.io/otel/trace"
 )
 
 // userHZ is the kernel jiffy rate exposed in /proc/<pid>/stat. USER_HZ is 100
 // on every Linux architecture spinifex targets (x86_64, aarch64).
 const userHZ = 100.0
 
+// pollerTracerName is the instrumentation scope for per-instance poll spans.
+const pollerTracerName = "github.com/mulgadc/spinifex/spinifex/services/qmpcollector"
+
 // sample is one raw counter snapshot; published values are deltas between
 // consecutive samples.
 type sample struct {
@@ -94,7 +101,7 @@ func (p *poller) run(ctx context.Context) {
 	ticker := time.NewTicker(period)
 	defer ticker.Stop()
 	for {
-		p.tick()
+		p.tick(ctx)
 		select {
 		case <-ctx.Done():
 			return
@@ -109,13 +116,20 @@ func (p *poller) run(ctx context.Context) {
 
 // tick takes a snapshot and publishes the delta against the previous one.
 // The first snapshot only primes counters; any error or counter regression
-// (QEMU restarted) re-primes instead of publishing garbage.
-func (p *poller) tick() {
+// (QEMU restarted) re-primes instead of publishing garbage. Each call opens a
+// root span so poll failures surface in APM alongside the journald warning.
+func (p *poller) tick(ctx context.Context) {
 	meta := p.snapshotMeta()
+	ctx, span := otel.Tracer(pollerTracerName).Start(ctx, "qmp.poll_instance",
+		trace.WithSpanKind(trace.SpanKindInternal),
+		trace.WithAttributes(attribute.String("instance.id", meta.InstanceID)))
+	defer span.End()
+
 	cur, err := p.collect(meta)
 	if err != nil {
-		slog.Warn("qmp-collector: collect failed, re-priming",
+		slog.WarnContext(ctx, "qmp-collector: collect failed, re-priming",
 			"instanceId", meta.InstanceID, "err", err)
+		utils.MarkSpanError(span, err)
 		p.mu.Lock()
 		p.prev = nil
 		p.mu.Unlock()
@@ -136,14 +150,16 @@ func (p *poller) tick() {
 	}
 	data, err := json.Marshal(batch)
 	if err != nil {
-		slog.Error("qmp-collector: marshal batch", "instanceId", meta.InstanceID, "err", err)
+		slog.ErrorContext(ctx, "qmp-collector: marshal batch", "instanceId", meta.InstanceID, "err", err)
+		utils.MarkSpanError(span, err)
 		return
 	}
 	if err := p.nc.Publish(types.MetricsEC2SubjectPrefix+meta.InstanceID, data); err != nil {
-		slog.Warn("qmp-collector: publish failed", "instanceId", meta.InstanceID, "err", err)
+		slog.WarnContext(ctx, "qmp-collector: publish failed", "instanceId", meta.InstanceID, "err", err)
+		utils.MarkSpanError(span, err)
 		return
 	}
-	slog.Debug("qmp-collector: published", "instanceId", meta.InstanceID, "series", len(batch.Series))
+	slog.DebugContext(ctx, "qmp-collector: published", "instanceId", meta.InstanceID, "series", len(batch.Series))
 }
 
 // collect dials the telemetry socket fresh each tick (no held connection to

spinifex/services/spinifexui/proxy.go

@@ -12,6 +12,8 @@ import (
 	"os"
 	"strings"
 	"time"
+
+	"go.opentelemetry.io/contrib/instrumentation/net/http/otelhttp"
 )
 
 // newProxyTransport creates an *http.Transport that trusts the given CA
@@ -40,7 +42,10 @@ func newProxyTransport(caCertPath string) (*http.Transport, error) {
 
 // newReverseProxy forwards requests to backendHost after stripping pathPrefix.
 // Sets req.Host to the backend so SigV4 canonical-host verification passes.
-func newReverseProxy(backendHost, pathPrefix string, transport *http.Transport) http.Handler {
+// The transport is wrapped with otelhttp so proxied requests inject
+// traceparent, letting a UI-initiated trace continue into awsgw/predastore
+// instead of stopping at spinifex-ui.
+func newReverseProxy(backendHost, pathPrefix string, transport http.RoundTripper) http.Handler {
 	target := &url.URL{
 		Scheme: "https",
 		Host:   backendHost,
@@ -57,7 +62,7 @@ func newReverseProxy(backendHost, pathPrefix string, transport *http.Transport)
 				pr.Out.URL.Path = "/"
 			}
 		},
-		Transport: transport,
+		Transport: otelhttp.NewTransport(transport),
 		ErrorHandler: func(w http.ResponseWriter, r *http.Request, err error) {
 			slog.Error("Proxy error", "backend", backendHost, "path", r.URL.Path, "error", err)
 			w.Header().Set("Content-Type", "application/xml")

spinifex/services/spinifexui/spinifexui.go

@@ -19,6 +19,7 @@ import (
 
 	"github.com/go-chi/chi/v5/middleware"
 	"github.com/mulgadc/spinifex/internal/tlsconfig"
+	"github.com/mulgadc/spinifex/spinifex/otelsetup"
 	"github.com/mulgadc/spinifex/spinifex/utils"
 )
 
@@ -219,7 +220,8 @@ func (svc *Service) launchService() error {
 	compressor := middleware.NewCompressor(5, "text/html", "text/css",
 		"application/javascript", "text/javascript", "application/json",
 		"image/svg+xml", "text/plain")
-	finalHandler := securityHeadersMiddleware(compressor.Handler(mux))
+	traced := otelsetup.HTTPMiddleware("spinifex-ui")(mux)
+	finalHandler := securityHeadersMiddleware(compressor.Handler(traced))
 
 	addr := fmt.Sprintf("%s:%d", svc.Config.Host, svc.Config.Port)