mirror of
https://github.com/mulgadc/spinifex.git
synced 2026-10-15 17:11:30 +00:00
fix: log every duration as integer milliseconds under a _ms key (#827)
slog renders a bare time.Duration as an int64 nanosecond count and names the unit nowhere, so every duration field in the fleet read like "elapsed":135000000000. The repo idiom of .Round(time.Second) changed the value without changing the shape and truncated every sub-second measurement to zero. otelsetup.Millis converts a duration for logging under a snake_case key ending in _ms, and TestSlogDurationConvention type-checks ./spinifex/... and ./cmd/... to fail the build when a raw Duration reaches an slog call or a Millis value is logged under a key without the suffix. It found 55 sites across 34 files, well past the sites a grep for time.Since shows, because most durations come from a config field or a package constant. A handler-level ReplaceAttr would fix every site at once but can only emit d.String(), which Elasticsearch maps as a keyword and nothing can aggregate over.
This commit is contained in:
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
})
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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
|
||||
|
||||
+11
-11
@@ -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.
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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) {
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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():
|
||||
|
||||
@@ -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():
|
||||
|
||||
@@ -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),
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
}
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
}
|
||||
|
||||
|
||||
+2
-1
@@ -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")
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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() }
|
||||
@@ -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 \"<name>%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
|
||||
}
|
||||
}
|
||||
@@ -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
|
||||
}
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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() {
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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")
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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():
|
||||
|
||||
@@ -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())
|
||||
|
||||
Reference in New Issue
Block a user