Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
9 changes: 9 additions & 0 deletions packages/api/internal/sandbox/storage/redis/heal.go
Original file line number Diff line number Diff line change
Expand Up @@ -100,6 +100,15 @@ func (s *Storage) healExpirationIndex(ctx context.Context) (int, teamWalk, error
}
walk.complete = skipped == 0

// Sample the ZSET cardinality once per heal pass. A single ZCARD piggybacked
// here costs one Redis round-trip every 5 min — negligible next to the
// forEachSandboxBatch scan that precedes it.
if n, zErr := s.redisClient.ZCard(ctx, globalExpirationSet).Result(); zErr == nil {
s.metrics.indexSize.Record(ctx, n)
} else {
logger.L().Warn(ctx, "Failed to sample expiration index size", zap.Error(zErr))
}

return healed, walk, nil
}

Expand Down
7 changes: 6 additions & 1 deletion packages/api/internal/sandbox/storage/redis/items.go
Original file line number Diff line number Diff line change
Expand Up @@ -55,7 +55,12 @@ func membersOf(refs []memberRef) []any {
// ExpiredItems returns running sandboxes whose EndTime has passed.
// It bounds per-cycle work via LIMIT and cleans up orphaned ZSET entries.
func (s *Storage) ExpiredItems(ctx context.Context) ([]sandboxtypes.Sandbox, error) {
now := time.Now()
sweepStart := time.Now()
defer func() {
s.metrics.sweepDuration.Record(ctx, time.Since(sweepStart).Milliseconds())
}()

now := sweepStart
nowMs := float64(now.UnixMilli())

// Fetch members whose score (EndTime in ms) is <= now, bounded to 256 per cycle.
Expand Down
17 changes: 17 additions & 0 deletions packages/api/internal/sandbox/storage/redis/main.go
Original file line number Diff line number Diff line change
Expand Up @@ -84,6 +84,11 @@ type expirationIndexMetrics struct {
sweptOrphan metric.MeasurementOption
sweptDeadExecution metric.MeasurementOption
sweptInvalid metric.MeasurementOption

// indexSize records the ZSET cardinality once per heal pass (every 5 min).
indexSize metric.Int64Histogram
// sweepDuration records wall-clock time of each ExpiredItems call.
sweepDuration metric.Int64Histogram
}

const sweptReasonAttr = "reason"
Expand All @@ -109,6 +114,16 @@ func newExpirationIndexMetrics(meter metric.Meter) (expirationIndexMetrics, erro
return expirationIndexMetrics{}, fmt.Errorf("expiration index swept counter: %w", err)
}

indexSize, err := telemetry.GetHistogram(meter, telemetry.ApiRedisStorageExpirationIndexSize)
if err != nil {
return expirationIndexMetrics{}, fmt.Errorf("expiration index size histogram: %w", err)
}

sweepDuration, err := telemetry.GetHistogram(meter, telemetry.ApiRedisStorageExpirationIndexSweepDuration)
if err != nil {
return expirationIndexMetrics{}, fmt.Errorf("expiration index sweep duration histogram: %w", err)
}

return expirationIndexMetrics{
indexHealed: healed,
indexRescored: rescored,
Expand All @@ -117,6 +132,8 @@ func newExpirationIndexMetrics(meter metric.Meter) (expirationIndexMetrics, erro
sweptOrphan: metric.WithAttributeSet(attribute.NewSet(attribute.String(sweptReasonAttr, "orphan"))),
sweptDeadExecution: metric.WithAttributeSet(attribute.NewSet(attribute.String(sweptReasonAttr, "dead_execution"))),
sweptInvalid: metric.WithAttributeSet(attribute.NewSet(attribute.String(sweptReasonAttr, "invalid"))),
indexSize: indexSize,
sweepDuration: sweepDuration,
}, nil
}

Expand Down
32 changes: 24 additions & 8 deletions packages/shared/pkg/telemetry/meters.go
Original file line number Diff line number Diff line change
Expand Up @@ -739,6 +739,18 @@ const (
)

const (
// ApiRedisStorageExpirationIndexSize records the cardinality of the global
// expiration ZSET (sandbox:storage:global:expiration) once per heal pass.
// Use the max/p99 to detect unbounded growth; a rising trend without a
// corresponding rise in live sandbox count indicates stale entries are
// accumulating faster than they are swept.
ApiRedisStorageExpirationIndexSize HistogramType = "api.redis_storage.expiration_index.size"

// ApiRedisStorageExpirationIndexSweepDuration records the wall-clock time
// of each ExpiredItems call (ZRANGEBYSCORE + MGET pipeline). Rising p99
// alongside a rising expiration_index.size confirms O(log N + K) growth.
ApiRedisStorageExpirationIndexSweepDuration HistogramType = "api.redis_storage.expiration_index.sweep_duration"

ApiRedisStoragePublisherPublishDuration HistogramType = "api.redis_storage.publisher.publish.duration"
// ApiRedisScriptDuration is the round trip of one Lua script run against
// the sandbox store, by script (see ApiRedisScriptCalls).
Expand Down Expand Up @@ -1299,10 +1311,12 @@ func GetGaugeInt(meter metric.Meter, name GaugeIntType) (metric.Int64ObservableG
}

var histogramDesc = map[HistogramType]string{
ApiRedisStoragePublisherPublishDuration: "Duration of a single Redis PUBLISH round-trip from the storage publisher",
ApiRedisScriptDuration: "Round trip of one Lua script run against the sandbox store, by script",
ApiSandboxListAge: "Age of a served sandbox list: the store sweep's duration for a read, the wait from the sweep's start for a joined (shared) read, the snapshot's lag for a hit",
ApiRedisStorageListGets: "Commands one bulk record read issued, by the sandbox list, the evictor or the node reconcile: one GET per record on the sandbox-tagged keys",
ApiRedisStorageExpirationIndexSize: "Cardinality of sandbox:storage:global:expiration ZSET at the time of each heal pass; use max/p99 to detect unbounded growth",
ApiRedisStorageExpirationIndexSweepDuration: "Wall-clock duration of one ExpiredItems call (ZRANGEBYSCORE + MGET pipeline); rising p99 alongside rising index size confirms O(log N) growth",
ApiRedisStoragePublisherPublishDuration: "Duration of a single Redis PUBLISH round-trip from the storage publisher",
ApiRedisScriptDuration: "Round trip of one Lua script run against the sandbox store, by script",
ApiSandboxListAge: "Age of a served sandbox list: the store sweep's duration for a read, the wait from the sweep's start for a joined (shared) read, the snapshot's lag for a hit",
ApiRedisStorageListGets: "Commands one bulk record read issued, by the sandbox list, the evictor or the node reconcile: one GET per record on the sandbox-tagged keys",

FPHRunDurationName: "Duration of a completed free-page-hinting drain or run, labeled by phase",
FPHStopDurationName: "Duration of the stop of an abandoned hinting cycle, acknowledgement wait included",
Expand Down Expand Up @@ -1398,10 +1412,12 @@ var histogramDesc = map[HistogramType]string{
}

var histogramUnits = map[HistogramType]string{
ApiRedisStoragePublisherPublishDuration: "ms",
ApiRedisScriptDuration: "ms",
ApiSandboxListAge: "ms",
ApiRedisStorageListGets: "{command}",
ApiRedisStorageExpirationIndexSize: "{entry}",
ApiRedisStorageExpirationIndexSweepDuration: "ms",
ApiRedisStoragePublisherPublishDuration: "ms",
ApiRedisScriptDuration: "ms",
ApiSandboxListAge: "ms",
ApiRedisStorageListGets: "{command}",

FPHRunDurationName: "ms",
FPHStopDurationName: "ms",
Expand Down