diff --git a/spinifex/services/qmpcollector/poller.go b/spinifex/services/qmpcollector/poller.go index c80b3403d..dc14406bb 100644 --- a/spinifex/services/qmpcollector/poller.go +++ b/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//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 diff --git a/spinifex/services/spinifexui/proxy.go b/spinifex/services/spinifexui/proxy.go index f4d031eb6..4fb28cc4a 100644 --- a/spinifex/services/spinifexui/proxy.go +++ b/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") diff --git a/spinifex/services/spinifexui/spinifexui.go b/spinifex/services/spinifexui/spinifexui.go index 4c797b2a9..811dc73ca 100644 --- a/spinifex/services/spinifexui/spinifexui.go +++ b/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)