diff --git a/cmd/rds-agent/agent.go b/cmd/rds-agent/agent.go index 4a4ffeba1..f0eeb212f 100644 --- a/cmd/rds-agent/agent.go +++ b/cmd/rds-agent/agent.go @@ -15,6 +15,7 @@ import ( "github.com/mulgadc/spinifex/internal/gwsign" "github.com/mulgadc/spinifex/internal/rdsgw" handlers_rds "github.com/mulgadc/spinifex/spinifex/handlers/rds" + "github.com/mulgadc/spinifex/spinifex/otelsetup" ) // Overridable via -ldflags "-X main.version=...". @@ -406,10 +407,10 @@ func retryObserved(ctx context.Context, what string, fn func(context.Context) er } if attempt >= retryErrorAttempt { slog.ErrorContext(ctx, "rds-agent: "+what+" still failing, retrying", - "attempt", attempt, "retryIn", delay, "err", err) + "attempt", attempt, "retry_in_ms", otelsetup.Millis(delay), "err", err) } else { slog.WarnContext(ctx, "rds-agent: "+what+" failed, retrying", - "attempt", attempt, "retryIn", delay, "err", err) + "attempt", attempt, "retry_in_ms", otelsetup.Millis(delay), "err", err) } select { diff --git a/cmd/rds-agent/paramrecovery.go b/cmd/rds-agent/paramrecovery.go index 1cb924ced..d19ce85f3 100644 --- a/cmd/rds-agent/paramrecovery.go +++ b/cmd/rds-agent/paramrecovery.go @@ -2,6 +2,7 @@ package main import ( "context" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "log/slog" "time" @@ -66,7 +67,7 @@ func (g *paramGuard) Run(ctx context.Context) { } message := handlers_rds.ParameterRollbackMessage - slog.Error("rds-agent: "+message, "after", g.after) + slog.Error("rds-agent: "+message, "after_ms", otelsetup.Millis(g.after)) if g.cp != nil { if _, err := g.cp.SubmitState(ctx, g.id, handlers_rds.EngineHealthUnhealthy, message); err != nil { slog.Error("rds-agent: reporting the parameter rollback failed", "err", err) diff --git a/cmd/rds-agent/quiesce.go b/cmd/rds-agent/quiesce.go index 9ceeb1fc0..c1b6791aa 100644 --- a/cmd/rds-agent/quiesce.go +++ b/cmd/rds-agent/quiesce.go @@ -6,6 +6,7 @@ import ( "context" "errors" "fmt" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "io" "log/slog" "os/exec" @@ -72,7 +73,7 @@ func (q *quiesceState) beginHoldLocked(label string, session engineSession, hold session: session, expiry: time.AfterFunc(hold, func() { q.expire(label) }), } - slog.Info("rds-agent: engine quiesced for backup", "label", label, "hold", hold) + slog.Info("rds-agent: engine quiesced for backup", "label", label, "hold_ms", otelsetup.Millis(hold)) } // Takes the hold off the engine, if one is still there, and disarms its diff --git a/cmd/rds-agent/register.go b/cmd/rds-agent/register.go index 15afb6668..e6f7f8871 100644 --- a/cmd/rds-agent/register.go +++ b/cmd/rds-agent/register.go @@ -2,6 +2,7 @@ package main import ( "context" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "log/slog" "time" ) @@ -20,7 +21,7 @@ func (a *Agent) register(ctx context.Context) error { a.hb.setInterval(time.Duration(out.HeartbeatIntervalSeconds) * time.Second) slog.Info("rds-agent: registered", - "dbInstanceIdentifier", a.id.DBInstanceIdentifier, "heartbeat", a.hb.interval) + "dbInstanceIdentifier", a.id.DBInstanceIdentifier, "heartbeat_ms", otelsetup.Millis(a.hb.interval)) return nil }) } diff --git a/cmd/spinifex/cmd/cluster.go b/cmd/spinifex/cmd/cluster.go index ad66e8c76..bdf69922c 100644 --- a/cmd/spinifex/cmd/cluster.go +++ b/cmd/spinifex/cmd/cluster.go @@ -12,6 +12,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/daemon" "github.com/mulgadc/spinifex/spinifex/network/external/dhcp" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "github.com/mulgadc/spinifex/spinifex/types" "github.com/nats-io/nats.go" "github.com/spf13/cobra" @@ -333,7 +334,7 @@ func runNodeDrainLocal(cmd *cobra.Command, args []string) { // `systemctl --failed`) without blocking the target's teardown; // exiting 0 would hide guests left running as storage vanishes. slog.Error("drain phase failed: no ACK from local daemon before timeout; guests may be left running while storage is about to be torn down", - "phase", phase, "node", node, "timeout", timeout, "error", err) + "phase", phase, "node", node, "timeout_ms", otelsetup.Millis(timeout), "error", err) reportGuestsLeftRunning(true, fmt.Sprintf("[%s] drain failed: guests left running with storage about to be torn down", strings.ToUpper(phase))) os.Exit(1) } diff --git a/go.mod b/go.mod index e4828a6c0..da606aee1 100644 --- a/go.mod +++ b/go.mod @@ -46,6 +46,7 @@ require ( golang.org/x/crypto v0.55.0 golang.org/x/sync v0.22.0 golang.org/x/sys v0.47.0 + golang.org/x/tools v0.48.0 gopkg.in/ini.v1 v1.67.3 ) @@ -202,7 +203,6 @@ require ( golang.org/x/term v0.45.0 // indirect golang.org/x/text v0.41.0 // indirect golang.org/x/time v0.15.0 // indirect - golang.org/x/tools v0.48.0 // indirect golang.org/x/vuln v1.6.0 // indirect google.golang.org/genproto/googleapis/api v0.0.0-20260803160001-6ac0973c030d // indirect google.golang.org/genproto/googleapis/rpc v0.0.0-20260803160001-6ac0973c030d // indirect diff --git a/spinifex/daemon/daemon.go b/spinifex/daemon/daemon.go index 786980ff6..e7e9d46da 100644 --- a/spinifex/daemon/daemon.go +++ b/spinifex/daemon/daemon.go @@ -1385,7 +1385,7 @@ func (d *Daemon) startLocal() error { d.shutdownWg.Go(d.monitorPeerReachability) d.ready.Store(true) - slog.Info("Daemon local-bootstrap complete", "node", d.node, "elapsed", time.Since(d.startTime).Round(time.Second)) + slog.Info("Daemon local-bootstrap complete", "node", d.node, "elapsed_ms", otelsetup.Millis(time.Since(d.startTime))) return nil } @@ -1473,7 +1473,7 @@ func (d *Daemon) startCluster() error { // flipping the prod default (which would re-introduce the SPOF that 1d // removed). if err := d.connectNATS(utils.WithMaxWait(d.requireNATSTimeout)); err != nil { - slog.Error("SPINIFEX_REQUIRE_NATS=1 set, NATS connect failed within 30s, aborting", "err", err, "timeout", d.requireNATSTimeout) + slog.Error("SPINIFEX_REQUIRE_NATS=1 set, NATS connect failed within 30s, aborting", "err", err, "timeout_ms", otelsetup.Millis(d.requireNATSTimeout)) d.exitFunc(1) return fmt.Errorf("connect NATS (strict): %w", err) } @@ -2018,7 +2018,7 @@ func (d *Daemon) startCluster() error { // No-op when northstar is not configured. if d.dnsReconciler.Enabled() { go d.dnsReconciler.Run(d.ctx) - slog.Info("Started DNS reconcile backstop", "interval", handlers_dns.DefaultReconcileInterval) + slog.Info("Started DNS reconcile backstop", "interval_ms", otelsetup.Millis(handlers_dns.DefaultReconcileInterval)) } // Initialize per-instance-type NATS subscriptions for capacity-aware routing. @@ -2066,7 +2066,7 @@ func (d *Daemon) startCluster() error { } d.ready.Store(true) - slog.Info("Daemon fully initialized", "node", d.node, "startupTime", time.Since(d.startTime).Round(time.Second)) + slog.Info("Daemon fully initialized", "node", d.node, "startup_time_ms", otelsetup.Millis(time.Since(d.startTime))) // Return once bootstrap is done. Start already installed the signal handler // and owns the single awaitShutdown; waiting here would block on the wait @@ -2216,7 +2216,7 @@ func (d *Daemon) initJetStream() error { if err == nil { d.jsManager.SetSyncObserver(d) - slog.Info("JetStream KV stores initialized successfully", "replicas", 1, "attempts", attempt, "elapsed", time.Since(start).Round(time.Second)) + slog.Info("JetStream KV stores initialized successfully", "replicas", 1, "attempts", attempt, "elapsed_ms", otelsetup.Millis(time.Since(start))) break } @@ -2225,7 +2225,7 @@ func (d *Daemon) initJetStream() error { return fmt.Errorf("failed to initialize JetStream after %s (%d attempts): %w", elapsed.Round(time.Second), attempt, err) } - slog.Warn("JetStream not ready (waiting for cluster quorum)", "error", err, "attempt", attempt, "elapsed", elapsed.Round(time.Second), "retryIn", retryDelay) + slog.Warn("JetStream not ready (waiting for cluster quorum)", "error", err, "attempt", attempt, "elapsed_ms", otelsetup.Millis(elapsed), "retry_in_ms", otelsetup.Millis(retryDelay)) time.Sleep(retryDelay) retryDelay = min(retryDelay*2, 10*time.Second) } @@ -2263,7 +2263,7 @@ func initServiceWithRetry[T any](name string, initFn func() (T, error)) (T, erro result, err := initFn() if err == nil { if attempt > 1 { - slog.Info(name+" initialized successfully", "attempts", attempt, "elapsed", time.Since(start).Round(time.Second)) + slog.Info(name+" initialized successfully", "attempts", attempt, "elapsed_ms", otelsetup.Millis(time.Since(start))) } return result, nil } @@ -2274,7 +2274,7 @@ func initServiceWithRetry[T any](name string, initFn func() (T, error)) (T, erro return zero, fmt.Errorf("%s unavailable after %s (%d attempts): %w", name, elapsed.Round(time.Second), attempt, err) } - slog.Warn("Failed to init "+name, "error", err, "attempt", attempt, "elapsed", elapsed.Round(time.Second)) + slog.Warn("Failed to init "+name, "error", err, "attempt", attempt, "elapsed_ms", otelsetup.Millis(elapsed)) initRetrySleep(retryDelay) retryDelay = min(retryDelay*2, 10*time.Second) } @@ -2305,15 +2305,15 @@ func (d *Daemon) waitForClusterReady() { } if ready { - slog.Info("Cluster readiness check passed", "elapsed", time.Since(start)) + slog.Info("Cluster readiness check passed", "elapsed_ms", otelsetup.Millis(time.Since(start))) return } - slog.Debug("Cluster not ready, waiting...", "reason", reason, "elapsed", time.Since(start)) + slog.Debug("Cluster not ready, waiting...", "reason", reason, "elapsed_ms", otelsetup.Millis(time.Since(start))) time.Sleep(interval) } - slog.Warn("Cluster readiness timeout, proceeding with recovery anyway", "maxWait", maxWait) + slog.Warn("Cluster readiness timeout, proceeding with recovery anyway", "max_wait_ms", otelsetup.Millis(maxWait)) } // checkViperblockReady reports whether viperblock is reachable via NATS. diff --git a/spinifex/daemon/heartbeat.go b/spinifex/daemon/heartbeat.go index d43467a1e..3433d81ad 100644 --- a/spinifex/daemon/heartbeat.go +++ b/spinifex/daemon/heartbeat.go @@ -1,6 +1,7 @@ package daemon import ( + "github.com/mulgadc/spinifex/spinifex/otelsetup" "log/slog" "time" ) @@ -34,7 +35,7 @@ func (d *Daemon) startHeartbeat() { } }() - slog.Info("Heartbeat started", "interval", heartbeatInterval) + slog.Info("Heartbeat started", "interval_ms", otelsetup.Millis(heartbeatInterval)) } // publishHeartbeat builds and writes a heartbeat entry to KV. diff --git a/spinifex/daemon/jetstream.go b/spinifex/daemon/jetstream.go index 39cd524db..f096d31f3 100644 --- a/spinifex/daemon/jetstream.go +++ b/spinifex/daemon/jetstream.go @@ -11,6 +11,7 @@ import ( "time" "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/mulgadc/spinifex/spinifex/vm" "github.com/nats-io/nats.go" @@ -644,7 +645,7 @@ func (m *JetStreamManager) WriteStateBytesBestEffort(nodeID string, jsonData []b m.obs.RecordKVSyncFailure(InstanceStateBucket, err) } if errors.Is(err, context.DeadlineExceeded) { - slog.Warn("KV sync timed out (best-effort)", "key", key, "timeout", timeout) + slog.Warn("KV sync timed out (best-effort)", "key", key, "timeout_ms", otelsetup.Millis(timeout)) return } slog.Warn("KV sync failed (best-effort)", "key", key, "err", err) diff --git a/spinifex/daemon/ochre_deps.go b/spinifex/daemon/ochre_deps.go index 0a6491abb..9e23f623c 100644 --- a/spinifex/daemon/ochre_deps.go +++ b/spinifex/daemon/ochre_deps.go @@ -14,6 +14,7 @@ import ( handlers_ochrevector "github.com/mulgadc/spinifex/spinifex/handlers/ochrevector" "github.com/mulgadc/spinifex/spinifex/network/host" "github.com/mulgadc/spinifex/spinifex/objectstore" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "github.com/nats-io/nats.go/jetstream" ) @@ -228,7 +229,7 @@ func (d *Daemon) connectOchreAppliance(appliance *handlers_ochrevector.Appliance func(attempt int, backoff time.Duration, err error) { if attempt <= ochreStartupLogAttempts || attempt%ochreStartupLogEvery == 0 { slog.Warn("Ochre vector store: appliance not reachable, will keep retrying", - "attempt", attempt, "backoff", backoff, "err", err) + "attempt", attempt, "backoff_ms", otelsetup.Millis(backoff), "err", err) } }, func() (handlers_ochrevector.VectorBackend, error) { diff --git a/spinifex/gateway/auth.go b/spinifex/gateway/auth.go index 6479e3af7..974f6cfd6 100644 --- a/spinifex/gateway/auth.go +++ b/spinifex/gateway/auth.go @@ -14,6 +14,7 @@ import ( "github.com/mulgadc/bluebottle/pkg/sigv4" "github.com/mulgadc/spinifex/spinifex/awserrors" handlers_iam "github.com/mulgadc/spinifex/spinifex/handlers/iam" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "github.com/mulgadc/spinifex/spinifex/utils" ) @@ -62,7 +63,7 @@ func (gw *GatewayConfig) SigV4AuthMiddleware() func(http.Handler) http.Handler { "sourceIP", clientIP, "requestTime", signingTime(r), "serverTime", time.Now().UTC().Format("20060102T150405Z"), - "maxSkew", sigv4.MaxClockSkew) + "max_skew_ms", otelsetup.Millis(sigv4.MaxClockSkew)) // Anonymous: the request was rejected before its key id was parsed, // so there is no client identity for the lockout to protect. gw.RateLimiter.RecordFailure(clientIP, anonymousAttempt) diff --git a/spinifex/gateway/ecr/lifecycle_sweep.go b/spinifex/gateway/ecr/lifecycle_sweep.go index da763cd91..4451575bf 100644 --- a/spinifex/gateway/ecr/lifecycle_sweep.go +++ b/spinifex/gateway/ecr/lifecycle_sweep.go @@ -7,6 +7,7 @@ import ( "time" "github.com/mulgadc/spinifex/spinifex/handlers/ecr" + "github.com/mulgadc/spinifex/spinifex/otelsetup" ) // DefaultLifecycleSweepInterval is the period between lifecycle expiry cycles. @@ -48,7 +49,7 @@ func (s *LifecycleSweeper) Run(ctx context.Context) { ticker := time.NewTicker(s.interval) defer ticker.Stop() - slog.InfoContext(ctx, "ECR lifecycle sweeper started", "interval", s.interval) + slog.InfoContext(ctx, "ECR lifecycle sweeper started", "interval_ms", otelsetup.Millis(s.interval)) for { select { case <-ctx.Done(): diff --git a/spinifex/gateway/ecrauth/rotator.go b/spinifex/gateway/ecrauth/rotator.go index c3b65b74d..65377a45e 100644 --- a/spinifex/gateway/ecrauth/rotator.go +++ b/spinifex/gateway/ecrauth/rotator.go @@ -2,6 +2,7 @@ package gateway_ecrauth import ( "context" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "log/slog" "sort" "time" @@ -102,7 +103,7 @@ func (r *Rotator) Run(ctx context.Context) { ticker := time.NewTicker(r.interval) defer ticker.Stop() - slog.Info("ECR signing-key rotator started", "interval", r.interval, "rotateAfter", r.params.rotateAfter) + slog.Info("ECR signing-key rotator started", "interval_ms", otelsetup.Millis(r.interval), "rotate_after_ms", otelsetup.Millis(r.params.rotateAfter)) for { select { case <-ctx.Done(): diff --git a/spinifex/gateway/ratelimit.go b/spinifex/gateway/ratelimit.go index 9f850b49d..76f850add 100644 --- a/spinifex/gateway/ratelimit.go +++ b/spinifex/gateway/ratelimit.go @@ -9,6 +9,7 @@ import ( "time" "github.com/mulgadc/spinifex/spinifex/awserrors" + "github.com/mulgadc/spinifex/spinifex/otelsetup" ) const ( @@ -128,7 +129,7 @@ func (rl *AuthRateLimiter) RecordFailure(ip, fingerprint string) { slog.Warn("Rate limit: IP locked out", "ip", ip, "distinct_attempts", maxFailures, - "lockout_duration", lockout, + "lockout_ms", otelsetup.Millis(lockout), ) } } diff --git a/spinifex/handlers/acm/renewal.go b/spinifex/handlers/acm/renewal.go index 3c92be453..d933633b2 100644 --- a/spinifex/handlers/acm/renewal.go +++ b/spinifex/handlers/acm/renewal.go @@ -19,6 +19,7 @@ import ( "github.com/aws/aws-sdk-go/service/acm" "github.com/mulgadc/spinifex/spinifex/awserrors" + "github.com/mulgadc/spinifex/spinifex/otelsetup" ) const ( @@ -189,7 +190,7 @@ func (w *Worker) Run(ctx context.Context) { // A scan with nothing due logs nothing, so without this an operator cannot // distinguish a running worker from one that never started. - slog.InfoContext(ctx, "ACM renewal worker started", "scan_interval", w.scanInterval) + slog.InfoContext(ctx, "ACM renewal worker started", "scan_interval_ms", otelsetup.Millis(w.scanInterval)) w.scanOnce(ctx) // don't wait a full interval before the first pass for { diff --git a/spinifex/handlers/bedrock/reaper.go b/spinifex/handlers/bedrock/reaper.go index 076c8ab49..cdf819d07 100644 --- a/spinifex/handlers/bedrock/reaper.go +++ b/spinifex/handlers/bedrock/reaper.go @@ -8,6 +8,7 @@ import ( "sync" "time" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go/jetstream" ) @@ -237,12 +238,9 @@ func (r *Reaper) sweepEndpoint(ctx context.Context, kv jetstream.KeyValue, rec E } if r.shouldReap(updated, now) { - // Seconds as a plain number, not a time.Duration: slog renders a Duration - // as int64 nanoseconds in JSON, which no reader parses at a glance and - // which reaches the metrics sink in a unit nothing charts. slog.InfoContext(ctx, "bedrock reaper: reclaiming an idle endpoint", "model", rec.ModelID, "instanceId", rec.InstanceID, - "idleSeconds", int64(now.Sub(updated.LastActive()).Seconds())) + "idle_ms", otelsetup.Millis(now.Sub(updated.LastActive()))) if _, err := r.svc.Delete(ctx, &DeleteEndpointInput{ModelID: rec.ModelID}, utils.GlobalAccountID); err != nil { return fmt.Errorf("reap idle endpoint: %w", err) } diff --git a/spinifex/handlers/dns/zonelock.go b/spinifex/handlers/dns/zonelock.go index 2e2db80d7..1fdf36508 100644 --- a/spinifex/handlers/dns/zonelock.go +++ b/spinifex/handlers/dns/zonelock.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "log/slog" "math/rand/v2" "strings" @@ -151,7 +152,7 @@ func (z *zoneLock) Release(ctx context.Context) { defer cancel() if err := z.kv.Delete(releaseCtx, z.key); err != nil { slog.Warn("dns writer: failed to release zone lock (TTL will reap)", - "zone", z.zone, "holder", z.locker.holder, "ttl", zoneLockTTL, "error", err) + "zone", z.zone, "holder", z.locker.holder, "ttl_ms", otelsetup.Millis(zoneLockTTL), "error", err) } } diff --git a/spinifex/handlers/ecs/store.go b/spinifex/handlers/ecs/store.go index bb9a98c39..c6b43f026 100644 --- a/spinifex/handlers/ecs/store.go +++ b/spinifex/handlers/ecs/store.go @@ -9,6 +9,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/kvutil" "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" ) @@ -237,6 +238,6 @@ func InitLeaderBucket(ctx context.Context, js jetstream.JetStream) (jetstream.Ke if err := migrate.DefaultRegistry.RunKV(ctx, KVBucketECSLeader, kv, KVBucketECSLeaderVersion); err != nil { return nil, fmt.Errorf("migrate %s: %w", KVBucketECSLeader, err) } - slog.Info("ECS leader bucket initialized", "bucket", KVBucketECSLeader, "ttl", KVBucketECSLeaderTTL) + slog.Info("ECS leader bucket initialized", "bucket", KVBucketECSLeader, "ttl_ms", otelsetup.Millis(KVBucketECSLeaderTTL)) return kv, nil } diff --git a/spinifex/handlers/eks/cluster_reconciler.go b/spinifex/handlers/eks/cluster_reconciler.go index 2f1dbce5c..8059e219a 100644 --- a/spinifex/handlers/eks/cluster_reconciler.go +++ b/spinifex/handlers/eks/cluster_reconciler.go @@ -5,6 +5,7 @@ import ( "crypto/tls" "errors" "fmt" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "log/slog" "net/http" "strings" @@ -1009,7 +1010,7 @@ func (r *ClusterReconciler) failIfCreateTimedOut(ctx context.Context, meta *Clus return fmt.Errorf("mark create-timeout failed: %w", err) } slog.Warn("ClusterReconciler: CREATING timed out, marked FAILED", - "cluster", r.clusterName, "timeout", r.createTimeout, "reason", reason) + "cluster", r.clusterName, "timeout_ms", otelsetup.Millis(r.createTimeout), "reason", reason) return ErrReconcilerClusterFailed } diff --git a/spinifex/handlers/eks/store.go b/spinifex/handlers/eks/store.go index a3588bdbf..aadce170c 100644 --- a/spinifex/handlers/eks/store.go +++ b/spinifex/handlers/eks/store.go @@ -12,6 +12,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/kvutil" "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" ) @@ -246,6 +247,6 @@ func InitLeaderBucket(ctx context.Context, js jetstream.JetStream, replicas int) if err := migrate.DefaultRegistry.RunKV(ctx, KVBucketEKSLeader, kv, KVBucketEKSLeaderVersion); err != nil { return nil, fmt.Errorf("migrate %s: %w", KVBucketEKSLeader, err) } - slog.Info("EKS leader bucket initialized", "bucket", KVBucketEKSLeader, "ttl", KVBucketEKSLeaderTTL) + slog.Info("EKS leader bucket initialized", "bucket", KVBucketEKSLeader, "ttl_ms", otelsetup.Millis(KVBucketEKSLeaderTTL)) return kv, nil } diff --git a/spinifex/handlers/iam/retry.go b/spinifex/handlers/iam/retry.go index b7d3243d0..b76b6b143 100644 --- a/spinifex/handlers/iam/retry.go +++ b/spinifex/handlers/iam/retry.go @@ -3,6 +3,7 @@ package handlers_iam import ( "context" "fmt" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "log/slog" "time" @@ -26,7 +27,7 @@ func NewIAMServiceWithRetry(ctx context.Context, natsConn *nats.Conn, masterKey svc, err := NewIAMServiceImpl(ctx, natsConn, masterKey, clusterSize) if err == nil { if attempt > 1 { - slog.Info("IAM service initialized after retry", "attempts", attempt, "elapsed", time.Since(start).Round(time.Second)) + slog.Info("IAM service initialized after retry", "attempts", attempt, "elapsed_ms", otelsetup.Millis(time.Since(start))) } return svc, nil } @@ -36,7 +37,7 @@ func NewIAMServiceWithRetry(ctx context.Context, natsConn *nats.Conn, masterKey return nil, fmt.Errorf("IAM service unavailable after %s (%d attempts): %w", elapsed.Round(time.Second), attempt, err) } - slog.Warn("IAM service not ready (waiting for JetStream cluster quorum)", "error", err, "attempt", attempt, "elapsed", elapsed.Round(time.Second), "retryIn", retryDelay) + slog.Warn("IAM service not ready (waiting for JetStream cluster quorum)", "error", err, "attempt", attempt, "elapsed_ms", otelsetup.Millis(elapsed), "retry_in_ms", otelsetup.Millis(retryDelay)) time.Sleep(retryDelay) retryDelay = min(retryDelay*2, 10*time.Second) } diff --git a/spinifex/handlers/sts/session_credentials.go b/spinifex/handlers/sts/session_credentials.go index be0445ebb..80a94ec90 100644 --- a/spinifex/handlers/sts/session_credentials.go +++ b/spinifex/handlers/sts/session_credentials.go @@ -13,6 +13,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/kvutil" "github.com/mulgadc/spinifex/spinifex/migrate" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go/jetstream" ) @@ -141,8 +142,8 @@ func (s *STSServiceImpl) RunJanitor(ctx context.Context) { defer ticker.Stop() slog.Info("STS session credential janitor started", - "interval", janitorInterval, - "grace_period", janitorGracePeriod) + "interval_ms", otelsetup.Millis(janitorInterval), + "grace_period_ms", otelsetup.Millis(janitorGracePeriod)) for { select { diff --git a/spinifex/lbagent/agent.go b/spinifex/lbagent/agent.go index fb4f2a0d0..c16cd2109 100644 --- a/spinifex/lbagent/agent.go +++ b/spinifex/lbagent/agent.go @@ -27,6 +27,7 @@ import ( "github.com/mulgadc/spinifex/internal/gwsign" "github.com/mulgadc/spinifex/internal/tlsconfig" "github.com/mulgadc/spinifex/spinifex/awserrors" + "github.com/mulgadc/spinifex/spinifex/otelsetup" ) const ( @@ -219,7 +220,7 @@ func (a *Agent) tick() { // logs at ERROR so the console telemetry still catches a blackhole. if !a.firstHeartbeatOK && time.Since(a.startedAt) < registrationGrace { slog.Info("Heartbeat pending during startup", "err", err, - "gateway", a.gatewayURL, "elapsed", time.Since(a.startedAt).Round(time.Second)) + "gateway", a.gatewayURL, "elapsed_ms", otelsetup.Millis(time.Since(a.startedAt))) return } diff --git a/spinifex/network/external/dhcp/manager.go b/spinifex/network/external/dhcp/manager.go index 9e8107bbf..92d280346 100644 --- a/spinifex/network/external/dhcp/manager.go +++ b/spinifex/network/external/dhcp/manager.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "log/slog" "math/rand/v2" "net" @@ -493,7 +494,7 @@ func (m *Manager) acquireWithBackoff(ctx context.Context, req AcquireRequest) (* lastErr = err slog.Warn("dhcp manager: acquire attempt failed", "client_id", req.ClientID, "attempt", i+1, "of", len(m.acquireSchedule), - "timeout", attempt, "err", err) + "timeout_ms", otelsetup.Millis(attempt), "err", err) } if lastErr == nil { lastErr = errors.New("acquire budget exhausted before first attempt") diff --git a/spinifex/network/host/firewall.go b/spinifex/network/host/firewall.go index 0b20931ea..f4d7aae50 100644 --- a/spinifex/network/host/firewall.go +++ b/spinifex/network/host/firewall.go @@ -16,6 +16,7 @@ import ( "time" "github.com/mulgadc/spinifex/spinifex/config" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "github.com/mulgadc/spinifex/spinifex/utils" ) @@ -88,7 +89,7 @@ func MaintainFirewall(ctx context.Context, configPath string, clusterConfig *con delay := firewallRetryDelay for { if err := ReconcileFirewall(configPath, clusterConfig); err != nil { - slog.Warn("Failed to reconcile host firewall, will retry", "err", err, "retry_in", delay) + slog.Warn("Failed to reconcile host firewall, will retry", "err", err, "retry_in_ms", otelsetup.Millis(delay)) delay = min(delay*2, firewallReconcileInterval) } else { delay = firewallReconcileInterval diff --git a/spinifex/network/reconcile/apply.go b/spinifex/network/reconcile/apply.go index f7bf0226d..6c5d9c484 100644 --- a/spinifex/network/reconcile/apply.go +++ b/spinifex/network/reconcile/apply.go @@ -12,6 +12,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/network/ovn/nbdb" "github.com/mulgadc/spinifex/spinifex/network/policy" "github.com/mulgadc/spinifex/spinifex/network/topology" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "github.com/mulgadc/spinifex/spinifex/utils" ) @@ -405,7 +406,7 @@ func (r *reconciler) ensureGatewayDatapath(ctx context.Context, vpcID, gwIP, eip } if time.Now().After(deadline) { slog.Error("reconcile/apply: gateway datapath did not recover after uplink repair; external connectivity degraded", - "vpc_id", vpcID, "target", target, "timeout", gatewayDatapathTimeout) + "vpc_id", vpcID, "target", target, "timeout_ms", otelsetup.Millis(gatewayDatapathTimeout)) return } select { @@ -476,7 +477,7 @@ func (r *reconciler) ensureGatewayClaimed(ctx context.Context, crPortName string } if time.Now().After(deadline) { slog.Error("reconcile/apply: gateway SB chassis claim did not converge; floating IPs may be unreachable", - "port", crPortName, "timeout", gatewayClaimTimeout) + "port", crPortName, "timeout_ms", otelsetup.Millis(gatewayClaimTimeout)) return } select { @@ -674,7 +675,7 @@ func (r *reconciler) ensureGuestPortDatapath(ctx context.Context, vpcID, lspName } if time.Now().After(deadline) { slog.Error("reconcile/apply: guest port datapath did not converge; EIP ingress may be unreachable", - "vpc_id", vpcID, "lsp", lspName, "timeout", guestPortDatapathTimeout) + "vpc_id", vpcID, "lsp", lspName, "timeout_ms", otelsetup.Millis(guestPortDatapathTimeout)) return } select { diff --git a/spinifex/network/reconcile/lock.go b/spinifex/network/reconcile/lock.go index 7e7e33f8d..9a35299e5 100644 --- a/spinifex/network/reconcile/lock.go +++ b/spinifex/network/reconcile/lock.go @@ -3,6 +3,7 @@ package reconcile import ( "context" "errors" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "log/slog" "time" @@ -58,7 +59,7 @@ func AcquireLeader(ctx context.Context, nc *nats.Conn, bucket, holder string) (f } if time.Now().After(deadline) { slog.Error("reconcile/lock: JetStream KV unreachable after retry, skipping reconcile", - "holder", holder, "bucket", bucket, "waited", leaderRetryFor, "err", err) + "holder", holder, "bucket", bucket, "waited_ms", otelsetup.Millis(leaderRetryFor), "err", err) return nil, false } slog.Debug("reconcile/lock: JetStream KV not ready, retrying", "holder", holder, "bucket", bucket, "err", err) diff --git a/spinifex/otelsetup/duration.go b/spinifex/otelsetup/duration.go new file mode 100644 index 000000000..33b055809 --- /dev/null +++ b/spinifex/otelsetup/duration.go @@ -0,0 +1,15 @@ +package otelsetup + +import "time" + +// DurationSuffix is the key suffix every logged duration carries. slog logs a +// bare time.Duration as an int64 count of nanoseconds and names the unit +// nowhere, so a reader cannot tell one duration field from another without the +// call site. The unit lives in the key instead. +const DurationSuffix = "_ms" + +// Millis converts d for logging under a key ending in DurationSuffix, e.g. +// slog.Info("done", "elapsed_ms", otelsetup.Millis(time.Since(start))). Never +// round the argument first: the value is already whole milliseconds, and +// rounding to a second logs 0 for everything faster than that. +func Millis(d time.Duration) int64 { return d.Milliseconds() } diff --git a/spinifex/otelsetup/duration_guard_test.go b/spinifex/otelsetup/duration_guard_test.go new file mode 100644 index 000000000..b416a8404 --- /dev/null +++ b/spinifex/otelsetup/duration_guard_test.go @@ -0,0 +1,164 @@ +package otelsetup_test + +import ( + "go/ast" + "go/token" + "go/types" + "os" + "path/filepath" + "strconv" + "strings" + "testing" + + "golang.org/x/tools/go/packages" + + "github.com/mulgadc/spinifex/spinifex/otelsetup" +) + +// logMethods are the slog entry points that take loosely typed key/value pairs, +// where a duration argument carries no unit of its own. +var logMethods = map[string]bool{ + "Debug": true, "DebugContext": true, + "Info": true, "InfoContext": true, + "Warn": true, "WarnContext": true, + "Error": true, "ErrorContext": true, + "Log": true, "LogAttrs": true, +} + +// TestSlogDurationConvention enforces the one duration convention: every +// duration logged goes through otelsetup.Millis under a key ending in "_ms". +// A bare time.Duration is logged as unitless nanoseconds, so without this two +// duration fields in the same index can differ by a factor of a billion with +// nothing in the record to say so. +func TestSlogDurationConvention(t *testing.T) { + root := moduleRoot(t) + + cfg := &packages.Config{ + Mode: packages.NeedName | packages.NeedSyntax | packages.NeedTypes | + packages.NeedTypesInfo | packages.NeedDeps | packages.NeedImports, + Dir: root, + } + pkgs, err := packages.Load(cfg, "./spinifex/...", "./cmd/...") + if err != nil { + t.Fatalf("load packages: %v", err) + } + + var raw, unkeyed []string + for _, pkg := range pkgs { + for _, file := range pkg.Syntax { + ast.Inspect(file, func(n ast.Node) bool { + call, ok := n.(*ast.CallExpr) + if !ok || !isLogCall(pkg.TypesInfo, call) { + return true + } + for i, arg := range call.Args { + switch { + case isDuration(pkg.TypesInfo.TypeOf(arg)): + raw = append(raw, site(root, pkg.Fset, arg.Pos())) + case isMillisCall(pkg.TypesInfo, arg) && !precededByUnitKey(call.Args, i): + unkeyed = append(unkeyed, site(root, pkg.Fset, arg.Pos())) + } + } + return true + }) + } + } + + if len(raw) > 0 { + t.Errorf("slog call sites pass a raw time.Duration:\n %s\n"+ + "Log it as \"%s\", otelsetup.Millis(d) so the field carries its unit.", + strings.Join(raw, "\n "), otelsetup.DurationSuffix) + } + if len(unkeyed) > 0 { + t.Errorf("otelsetup.Millis logged under a key not ending in %q:\n %s", + otelsetup.DurationSuffix, strings.Join(unkeyed, "\n ")) + } +} + +// isLogCall reports whether call is a slog logging call or slog.Duration, both +// of which would take a duration argument without recording its unit. +func isLogCall(info *types.Info, call *ast.CallExpr) bool { + sel, ok := call.Fun.(*ast.SelectorExpr) + if !ok { + return false + } + fn, ok := info.ObjectOf(sel.Sel).(*types.Func) + if !ok || fn.Pkg() == nil || fn.Pkg().Path() != "log/slog" { + return false + } + return logMethods[sel.Sel.Name] || sel.Sel.Name == "Duration" +} + +// isMillisCall reports whether expr is a call to otelsetup.Millis. +func isMillisCall(info *types.Info, expr ast.Expr) bool { + call, ok := expr.(*ast.CallExpr) + if !ok { + return false + } + sel, ok := call.Fun.(*ast.SelectorExpr) + if !ok { + return false + } + fn, ok := info.ObjectOf(sel.Sel).(*types.Func) + if !ok || fn.Pkg() == nil { + return false + } + return fn.Name() == "Millis" && + fn.Pkg().Path() == "github.com/mulgadc/spinifex/spinifex/otelsetup" +} + +// precededByUnitKey reports whether the argument at i is the value half of a +// key/value pair whose key literal ends in the duration suffix. +func precededByUnitKey(args []ast.Expr, i int) bool { + if i == 0 { + return false + } + lit, ok := args[i-1].(*ast.BasicLit) + if !ok || lit.Kind != token.STRING { + return false + } + key, err := strconv.Unquote(lit.Value) + if err != nil { + return false + } + return strings.HasSuffix(key, otelsetup.DurationSuffix) +} + +// isDuration reports whether t is exactly time.Duration. +func isDuration(t types.Type) bool { + named, ok := t.(*types.Named) + if !ok { + return false + } + obj := named.Obj() + return obj.Pkg() != nil && obj.Pkg().Path() == "time" && obj.Name() == "Duration" +} + +// site renders pos as a module-relative file:line. +func site(root string, fset *token.FileSet, pos token.Pos) string { + p := fset.Position(pos) + rel, err := filepath.Rel(root, p.Filename) + if err != nil { + rel = p.Filename + } + return rel + ":" + strconv.Itoa(p.Line) +} + +// moduleRoot walks up from the test's directory to the directory holding go.mod. +func moduleRoot(t *testing.T) string { + t.Helper() + dir, err := os.Getwd() + if err != nil { + t.Fatalf("getwd: %v", err) + } + for { + if _, err := os.Stat(filepath.Join(dir, "go.mod")); err == nil { + return dir + } + parent := filepath.Dir(dir) + if parent == dir { + t.Fatal("no go.mod above the test directory") + } + dir = parent + } +} diff --git a/spinifex/services/viperblockd/viperblockd.go b/spinifex/services/viperblockd/viperblockd.go index 300a9cef2..8ff7fe1f2 100644 --- a/spinifex/services/viperblockd/viperblockd.go +++ b/spinifex/services/viperblockd/viperblockd.go @@ -18,6 +18,7 @@ import ( "github.com/mulgadc/bluebottle/pkg/masterkey" "github.com/mulgadc/spinifex/spinifex/admin" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "github.com/mulgadc/spinifex/spinifex/types" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/mulgadc/viperblock/viperblock" @@ -67,7 +68,7 @@ func retryLoadState(volume string, attempts int, baseDelay time.Duration, sleep break } slog.Warn("LoadState transient failure, retrying", - "volume", volume, "attempt", attempt, "delay", delay, "err", err) + "volume", volume, "attempt", attempt, "delay_ms", otelsetup.Millis(delay), "err", err) sleep(delay) delay = delay * 3 / 2 } diff --git a/spinifex/services/viperblockd/volume_lease.go b/spinifex/services/viperblockd/volume_lease.go index 3ca6c60a4..5403797ef 100644 --- a/spinifex/services/viperblockd/volume_lease.go +++ b/spinifex/services/viperblockd/volume_lease.go @@ -5,6 +5,7 @@ import ( "encoding/json" "errors" "fmt" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "log/slog" "regexp" "sync" @@ -236,7 +237,7 @@ func (l *volumeLeases) release(ctx context.Context, lease *volumeLease) { } if err := l.kv.Delete(ctx, lease.key, jetstream.LastRevision(revision)); err != nil { - slog.Warn("volume lease: delete failed, entry will expire", "volume", lease.volume, "ttl", volumeLeaseTTL, "err", err) + slog.Warn("volume lease: delete failed, entry will expire", "volume", lease.volume, "ttl_ms", otelsetup.Millis(volumeLeaseTTL), "err", err) return } slog.Info("volume lease released", "volume", lease.volume, "generation", lease.generation) diff --git a/spinifex/utils/nats.go b/spinifex/utils/nats.go index a2cb3595e..bd1606191 100644 --- a/spinifex/utils/nats.go +++ b/spinifex/utils/nats.go @@ -13,6 +13,7 @@ import ( "github.com/mulgadc/spinifex/internal/tlsconfig" "github.com/mulgadc/spinifex/spinifex/awserrors" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "github.com/nats-io/nats.go" ) @@ -176,7 +177,7 @@ func ConnectNATSWithRetry(host, token, caCertPath string, opts ...RetryOption) ( nc, err := ConnectNATS(host, token, caCertPath, opts...) if err == nil { if time.Since(start) > time.Second { - slog.Info("NATS connection established", "elapsed", time.Since(start).Round(time.Second)) + slog.Info("NATS connection established", "elapsed_ms", otelsetup.Millis(time.Since(start))) } return nc, nil } @@ -198,11 +199,11 @@ func ConnectNATSWithRetry(host, token, caCertPath string, opts ...RetryOption) ( // Past the escalation threshold, promote from Warn to Error (rate-limited to 1/min). if attempt > natsRetryEscalateAttempt { if lastEscalatedLog.IsZero() || time.Since(lastEscalatedLog) >= time.Minute { - slog.Error("NATS still disconnected", "error", err, "disconnected_for", elapsed.Round(time.Second), "attempt", attempt) + slog.Error("NATS still disconnected", "error", err, "disconnected_for_ms", otelsetup.Millis(elapsed), "attempt", attempt) lastEscalatedLog = time.Now() } } else { - slog.Warn("NATS not ready, retrying...", "error", err, "elapsed", elapsed.Round(time.Second), "retryIn", cfg.retryDelay, "attempt", attempt) + slog.Warn("NATS not ready, retrying...", "error", err, "elapsed_ms", otelsetup.Millis(elapsed), "retry_in_ms", otelsetup.Millis(cfg.retryDelay), "attempt", attempt) } if cfg.ctx != nil { diff --git a/spinifex/utils/nats_test.go b/spinifex/utils/nats_test.go index 8e665fd3b..d328f5a2f 100644 --- a/spinifex/utils/nats_test.go +++ b/spinifex/utils/nats_test.go @@ -675,7 +675,7 @@ func TestConnectNATSWithRetry_LogEscalatesPastThreshold(t *testing.T) { assert.GreaterOrEqual(t, warnCount, 1, "expect warn logs for first 30 attempts") assert.GreaterOrEqual(t, errCount, 1, "expect at least one escalated error log past the threshold") assert.LessOrEqual(t, errCount, 2, "rate-limited to once per minute, so a sub-second test should see at most one or two") - assert.Contains(t, logs, "disconnected_for=", "escalated error should include disconnected_for") + assert.Contains(t, logs, "disconnected_for_ms=", "escalated error should include disconnected_for_ms") } // TestConnectNATSWithRetry_NoEscalation_BelowThreshold keeps the attempt count diff --git a/spinifex/vm/crash_recovery.go b/spinifex/vm/crash_recovery.go index 1ff07f09a..d2c54b3a2 100644 --- a/spinifex/vm/crash_recovery.go +++ b/spinifex/vm/crash_recovery.go @@ -4,6 +4,7 @@ import ( "context" "errors" "fmt" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "log/slog" "os" "os/exec" @@ -181,7 +182,7 @@ func (m *Manager) MaybeRestart(instance *VM) { slog.Error("Instance exceeded max restarts in window, leaving in error state", "instance", instance.ID, "crashes", crashCount, - "window", RestartWindow, + "window_ms", otelsetup.Millis(RestartWindow), "max", MaxRestartsInWindow) m.releaseGPUOnTerminalCrash(instance) return @@ -209,7 +210,7 @@ func (m *Manager) MaybeRestart(instance *VM) { slog.Info("Scheduling instance restart", "instance", instance.ID, - "delay", delay, + "delay_ms", otelsetup.Millis(delay), "restartCount", restartCount+1) restartAfterFunc(delay, func() { diff --git a/spinifex/vm/stuck_terminate_reaper.go b/spinifex/vm/stuck_terminate_reaper.go index 7a817f87b..4082641e2 100644 --- a/spinifex/vm/stuck_terminate_reaper.go +++ b/spinifex/vm/stuck_terminate_reaper.go @@ -2,6 +2,7 @@ package vm import ( "context" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "log/slog" "time" ) @@ -58,7 +59,7 @@ func (r *StuckTerminateReaper) Sweep(context.Context) (int, error) { } slog.Warn("vm/gc: force-completing terminate wedged past timeout", - "instanceId", v.ID, "shuttingDownFor", time.Since(v.ShuttingDownAt)) + "instanceId", v.ID, "shutting_down_for_ms", otelsetup.Millis(time.Since(v.ShuttingDownAt))) if err := r.m.forceFinalizeStuckTerminate(v); err != nil { slog.Error("vm/gc: failed to force-complete wedged terminate, will retry next sweep", "instanceId", v.ID, "err", err) diff --git a/spinifex/vm/teardown_reaper.go b/spinifex/vm/teardown_reaper.go index 9f99c69a6..fb6aceb3b 100644 --- a/spinifex/vm/teardown_reaper.go +++ b/spinifex/vm/teardown_reaper.go @@ -3,6 +3,7 @@ package vm import ( "context" "fmt" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "log/slog" "strings" "sync" @@ -89,7 +90,7 @@ func (r *TerminatedTeardownReaper) recordRetry(instanceID, dep string, state Tea } b.next = time.Now().Add(wait) slog.Warn("vm/gc: teardown dependent failed, backing off", - "instanceId", instanceID, "dependent", dep, "failures", b.failures, "retry_in", wait) + "instanceId", instanceID, "dependent", dep, "failures", b.failures, "retry_in_ms", otelsetup.Millis(wait)) } // forgetInstance drops an instance's backoff once its record is gone, so the diff --git a/spinifex/vm/watchdog.go b/spinifex/vm/watchdog.go index 65b9a493e..eefc9ab0f 100644 --- a/spinifex/vm/watchdog.go +++ b/spinifex/vm/watchdog.go @@ -2,6 +2,7 @@ package vm import ( "context" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "log/slog" "time" ) @@ -47,7 +48,7 @@ func (m *Manager) scanAndMarkStuckPending(now time.Time) { for _, instance := range stuck { slog.Warn("Instance stuck in pending, marking failed", "instanceId", instance.ID, "status", instance.Status, - "elapsed", now.Sub(*instance.Instance.LaunchTime)) + "elapsed_ms", otelsetup.Millis(now.Sub(*instance.Instance.LaunchTime))) m.MarkFailed(context.Background(), instance, "launch_timeout") } } diff --git a/spinifex/vpcd/ovn_controller_watchdog.go b/spinifex/vpcd/ovn_controller_watchdog.go index 8a04ec53f..7f45876f7 100644 --- a/spinifex/vpcd/ovn_controller_watchdog.go +++ b/spinifex/vpcd/ovn_controller_watchdog.go @@ -6,6 +6,7 @@ import ( "time" "github.com/mulgadc/spinifex/spinifex/network/host" + "github.com/mulgadc/spinifex/spinifex/otelsetup" ) // Compile-time check: the gateway-claim prober satisfies the watchdog's needs. @@ -78,7 +79,7 @@ func (w *ovnWatchdog) evaluate(ctx context.Context, now time.Time) bool { } slog.Warn("ovn-controller watchdog: SB wedge detected, resetting SB cluster state", - "status", status, "stale_for", now.Sub(w.nonConnectedSince).Round(time.Second)) + "status", status, "stale_for_ms", otelsetup.Millis(now.Sub(w.nonConnectedSince))) if err := w.prober.ResetSBClusterState(ctx); err != nil { slog.Warn("ovn-controller watchdog: sb-cluster-state-reset failed", "err", err) // Still record the attempt so the cooldown throttles retries. @@ -95,7 +96,7 @@ func runOVNControllerWatchdog(ctx context.Context, w *ovnWatchdog) { ticker := time.NewTicker(ovnWatchdogInterval) defer ticker.Stop() slog.Info("ovn-controller wedge watchdog started", - "interval", ovnWatchdogInterval, "stale_after", w.staleAfter, "cooldown", w.cooldown) + "interval_ms", otelsetup.Millis(ovnWatchdogInterval), "stale_after_ms", otelsetup.Millis(w.staleAfter), "cooldown_ms", otelsetup.Millis(w.cooldown)) for { select { case <-ctx.Done(): diff --git a/spinifex/vpcd/vpcd.go b/spinifex/vpcd/vpcd.go index b491b4973..5f3f17691 100644 --- a/spinifex/vpcd/vpcd.go +++ b/spinifex/vpcd/vpcd.go @@ -26,6 +26,7 @@ import ( "github.com/mulgadc/spinifex/spinifex/network/reconcile" "github.com/mulgadc/spinifex/spinifex/network/subscribers" "github.com/mulgadc/spinifex/spinifex/network/topology" + "github.com/mulgadc/spinifex/spinifex/otelsetup" "github.com/mulgadc/spinifex/spinifex/utils" "github.com/nats-io/nats.go" "github.com/nats-io/nats.go/jetstream" @@ -64,13 +65,13 @@ var waitForFlowsHV = func(nbAddr string) error { ) if out, err := cmd.CombinedOutput(); err != nil { slog.Warn("vpcd: OVN flows-ready barrier overran; continuing without confirmation", - "elapsed", time.Since(start), + "elapsed_ms", otelsetup.Millis(time.Since(start)), "err", err, "output", strings.TrimSpace(string(out)), ) return nil } - slog.Debug("vpcd: OVN flows-ready barrier complete", "elapsed", time.Since(start)) + slog.Debug("vpcd: OVN flows-ready barrier complete", "elapsed_ms", otelsetup.Millis(time.Since(start))) return nil } @@ -377,7 +378,7 @@ func resolveExternalCIDR(ctx context.Context, bridge string, timeout time.Durati bridge, timeout, attempt, err) } slog.Warn("vpcd: external CIDR not yet assigned, retrying", - "bridge", bridge, "err", err, "attempt", attempt, "retry_in", retryDelay) + "bridge", bridge, "err", err, "attempt", attempt, "retry_in_ms", otelsetup.Millis(retryDelay)) select { case <-ctx.Done(): return netip.Prefix{}, fmt.Errorf("external CIDR resolution cancelled: %w", ctx.Err())