vrr/Pulse
2
0
Fork 0
mirror of https://github.com/rcourtman/Pulse.git synced 2026-10-19 02:28:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 13

Defer metrics store startup maintenance

Browse source
This commit is contained in:
rcourtman 2026-04-01 13:29:39 +01:00
parent f5a878016c
commit 4e16f35347
5 changed files with 148 additions and 41 deletions
Download patch file Download diff file Expand all files Collapse all files

5
docs/release-control/v6/internal/subsystems/performance-and-scalability.md
View file

@ -603,3 +603,8 @@ That same metrics-store boundary also owns the persistent DB file path. The
runtime must normalize the owned metrics directory and resolve the selected DB
filename through the shared storage-path helper before it creates directories
or opens SQLite, instead of trusting raw caller-built paths.
That same metrics hot path must also keep startup maintenance off the
constructor critical path. `NewStore` may initialize schema and return a usable
store, but restart-time retention cleanup and one-time auto-vacuum migration
must run through the background worker so large historical databases do not
block application startup before the metrics store is even available.

72
pkg/metrics/store.go
View file

@ -108,6 +108,12 @@ type WriteMetric struct {
Tier Tier
}
type maintenanceRequest struct {
run func()
}
var startupMaintenanceHook func()
// Store provides persistent metrics storage
type Store struct {
db *pdb.InstrumentedDB
@ -118,11 +124,12 @@ type Store struct {
buffer []bufferedMetric
// Background workers
writeCh chan []bufferedMetric
stopCh chan struct{}
doneCh chan struct{}
stopOnce sync.Once
stopping atomic.Bool
writeCh chan []bufferedMetric
maintenanceCh chan maintenanceRequest
stopCh chan struct{}
doneCh chan struct{}
stopOnce sync.Once
stopping atomic.Bool
}
// NewStore creates a new metrics store with the given configuration
@ -162,12 +169,13 @@ func NewStore(config StoreConfig) (*Store, error) {
db := pdb.Wrap(rawDB, "metrics")
store := &Store{
db: db,
config: config,
buffer: make([]bufferedMetric, 0, config.WriteBufferSize),
writeCh: make(chan []bufferedMetric, 100), // Buffer for write batches
stopCh: make(chan struct{}),
doneCh: make(chan struct{}),
db: db,
config: config,
buffer: make([]bufferedMetric, 0, config.WriteBufferSize),
writeCh: make(chan []bufferedMetric, 100), // Buffer for write batches
maintenanceCh: make(chan maintenanceRequest, 1),
stopCh: make(chan struct{}),
doneCh: make(chan struct{}),
}
// Initialize schema
@ -177,18 +185,9 @@ func NewStore(config StoreConfig) (*Store, error) {
}
store.migrateLegacyHostResourceType()
// Clean up stale data from previous runs before starting the background worker.
// This prevents accumulation if Pulse was restarted before hourly retention ran.
// Runs BEFORE auto-vacuum migration so the VACUUM operates on a much smaller
// dataset (e.g. 60MB of live data instead of 5GB of stale + live).
store.runRetention()
// Migrate existing databases to incremental auto-vacuum. This is a one-time
// operation that restructures the file so deleted pages can be reclaimed.
store.migrateAutoVacuum()
// Start background workers
go store.backgroundWorker()
store.enqueueMaintenance(store.runStartupMaintenance)
log.Info().
Str("path", config.DBPath).
@ -515,6 +514,32 @@ func (s *Store) WriteBatchSync(metrics []WriteMetric) {
s.writeBatch(batch)
}
func (s *Store) enqueueMaintenance(run func()) {
if run == nil {
return
}
select {
case s.maintenanceCh <- maintenanceRequest{run: run}:
default:
log.Debug().Msg("Metrics maintenance queue full, skipping duplicate request")
}
}
func (s *Store) runStartupMaintenance() {
start := time.Now()
if startupMaintenanceHook != nil {
startupMaintenanceHook()
}
// Run retention before any deferred vacuum work so restart cleanup trims
// stale rows before SQLite potentially rewrites the file.
s.runRetention()
s.migrateAutoVacuum()
log.Info().Dur("duration", time.Since(start)).Msg("Deferred metrics startup maintenance completed")
}
// flush writes buffered metrics to the database (caller must hold bufferMu)
func (s *Store) flushLocked() {
if len(s.buffer) == 0 {
@ -1161,6 +1186,11 @@ func (s *Store) backgroundWorker() {
case batch := <-s.writeCh:
s.writeBatch(batch)
case maintenance := <-s.maintenanceCh:
if maintenance.run != nil {
maintenance.run()
}
case <-flushTicker.C:
s.Flush()

55
pkg/metrics/store_additional_test.go
View file

@ -20,7 +20,7 @@ func TestStoreWriteBatchSync(t *testing.T) {
}
defer store.Close()
ts := time.Unix(1000, 0)
ts := time.Now().UTC().Truncate(time.Second)
metrics := []WriteMetric{
{ResourceType: "vm", ResourceID: "v1", MetricType: "cpu", Value: 10.0, Timestamp: ts, Tier: TierRaw},
{ResourceType: "vm", ResourceID: "v1", MetricType: "mem", Value: 50.0, Timestamp: ts, Tier: TierRaw},
@ -212,6 +212,53 @@ func TestStoreFlushLockedLogsStructuredContextWhenWriteChannelFull(t *testing.T)
}
}
func TestNewStoreDefersStartupMaintenance(t *testing.T) {
previousHook := startupMaintenanceHook
defer func() {
startupMaintenanceHook = previousHook
}()
started := make(chan struct{})
release := make(chan struct{})
startupMaintenanceHook = func() {
close(started)
<-release
}
dir := t.TempDir()
cfg := DefaultConfig(dir)
cfg.FlushInterval = time.Hour
done := make(chan struct{})
var (
store *Store
err error
)
go func() {
store, err = NewStore(cfg)
close(done)
}()
select {
case <-done:
case <-time.After(200 * time.Millisecond):
t.Fatal("NewStore blocked on startup maintenance")
}
if err != nil {
t.Fatalf("NewStore returned error: %v", err)
}
select {
case <-started:
case <-time.After(time.Second):
t.Fatal("startup maintenance was not scheduled")
}
close(release)
defer store.Close()
}
func TestStoreMigratesLegacyHostResourceTypeToAgent(t *testing.T) {
dir := t.TempDir()
cfg := DefaultConfig(dir)
@ -373,7 +420,7 @@ func TestStoreQueryCanonicalizesIdentifiers(t *testing.T) {
t.Fatalf("NewStore returned error: %v", err)
}
defer store.Close()
ts := time.Unix(1710000000, 0)
ts := time.Now().UTC().Truncate(time.Second)
store.WriteBatchSync([]WriteMetric{{
ResourceType: "agent",
@ -409,7 +456,7 @@ func TestStoreCanonicalizesMetricTypeCaseOnWriteAndQuery(t *testing.T) {
}
defer store.Close()
ts := time.Unix(1710000200, 0)
ts := time.Now().UTC().Truncate(time.Second)
store.WriteBatchSync([]WriteMetric{{
ResourceType: "agent",
ResourceID: "agent-1",
@ -451,7 +498,7 @@ func TestStoreQueryAllBatchCanonicalizesAndDeduplicatesIdentifiers(t *testing.T)
t.Fatalf("NewStore returned error: %v", err)
}
defer store.Close()
ts := time.Unix(1710000100, 0)
ts := time.Now().UTC().Truncate(time.Second)
store.WriteBatchSync([]WriteMetric{
{ResourceType: "agent", ResourceID: "agent-1", MetricType: "cpu", Value: 11.0, Timestamp: ts, Tier: TierRaw},

45
pkg/metrics/store_downsample_test.go
View file

@ -6,6 +6,26 @@ import (
"time"
)
func recentBucketOffsetTime(stepSecs, offsetSecs int64) time.Time {
if stepSecs <= 0 {
return time.Now().UTC().Truncate(time.Second)
}
if offsetSecs < 0 {
offsetSecs = 0
}
if offsetSecs >= stepSecs {
offsetSecs = stepSecs - 1
}
now := time.Now().UTC()
bucketStart := (now.Unix() / stepSecs) * stepSecs
ts := time.Unix(bucketStart+offsetSecs, 0).UTC()
if ts.After(now) {
ts = ts.Add(-time.Duration(stepSecs) * time.Second)
}
return ts
}
func TestStoreQueryAllDownsampling(t *testing.T) {
dir := t.TempDir()
store, err := NewStore(DefaultConfig(dir))
@ -14,7 +34,7 @@ func TestStoreQueryAllDownsampling(t *testing.T) {
}
defer store.Close()
start := time.Unix(1000, 0)
start := recentBucketOffsetTime(300, 100)
batch := make([]bufferedMetric, 0, 20)
for i := 0; i < 10; i++ {
ts := start.Add(time.Duration(i) * time.Minute)
@ -52,13 +72,17 @@ func TestStoreQueryAllDownsampling(t *testing.T) {
}
}
assertPoint(cpu[0], 1050, 1.5, 0, 3)
assertPoint(cpu[1], 1350, 6, 4, 8)
assertPoint(cpu[2], 1650, 9, 9, 9)
bucketCenter := func(ts time.Time, step int64) int64 {
return ((ts.Unix() / step) * step) + (step / 2)
}
assertPoint(mem[0], 1050, 101.5, 100, 103)
assertPoint(mem[1], 1350, 106, 104, 108)
assertPoint(mem[2], 1650, 109, 109, 109)
assertPoint(cpu[0], bucketCenter(start, 300), 1.5, 0, 3)
assertPoint(cpu[1], bucketCenter(start.Add(5*time.Minute), 300), 6, 4, 8)
assertPoint(cpu[2], bucketCenter(start.Add(9*time.Minute), 300), 9, 9, 9)
assertPoint(mem[0], bucketCenter(start, 300), 101.5, 100, 103)
assertPoint(mem[1], bucketCenter(start.Add(5*time.Minute), 300), 106, 104, 108)
assertPoint(mem[2], bucketCenter(start.Add(9*time.Minute), 300), 109, 109, 109)
}
func TestStoreTierFallbacks(t *testing.T) {
@ -134,7 +158,7 @@ func TestStoreQueryDownsamplingStats(t *testing.T) {
}
defer store.Close()
start := time.Unix(1000, 0)
start := time.Now().UTC().Truncate(time.Minute)
store.writeBatch([]bufferedMetric{
{resourceType: "vm", resourceID: "v2", metricType: "cpu", value: 10, timestamp: start, tier: TierRaw},
{resourceType: "vm", resourceID: "v2", metricType: "cpu", value: 30, timestamp: start.Add(20 * time.Second), tier: TierRaw},
@ -150,8 +174,9 @@ func TestStoreQueryDownsamplingStats(t *testing.T) {
}
point := points[0]
if point.Timestamp.Unix() != 1020 {
t.Fatalf("expected bucket timestamp 1020, got %d", point.Timestamp.Unix())
expectedBucket := time.Unix(((start.Unix()/120)*120)+(120/2), 0)
if point.Timestamp.Unix() != expectedBucket.Unix() {
t.Fatalf("expected bucket timestamp %d, got %d", expectedBucket.Unix(), point.Timestamp.Unix())
}
if point.Value != 20 || point.Min != 10 || point.Max != 30 {
t.Fatalf("unexpected stats: value=%v min=%v max=%v", point.Value, point.Min, point.Max)

12
pkg/metrics/store_test.go
View file

@ -23,7 +23,7 @@ func TestStoreWriteBatchAndQuery(t *testing.T) {
}
defer store.Close()
ts := time.Unix(1000, 0)
ts := time.Now().UTC().Truncate(time.Second)
store.writeBatch([]bufferedMetric{
{resourceType: "vm", resourceID: "vm-101", metricType: "cpu", value: 1.5, timestamp: ts, tier: TierRaw},
{resourceType: "vm", resourceID: "vm-101", metricType: "cpu", value: 2.5, timestamp: ts.Add(1 * time.Second), tier: TierRaw},
@ -79,7 +79,7 @@ func TestStoreSelectTierAndStats(t *testing.T) {
}
// Insert one point for each tier to verify stats aggregation.
ts := int64(1000)
ts := time.Now().UTC().Unix()
_, err = store.db.Exec(
`INSERT INTO metrics (resource_type, resource_id, metric_type, value, timestamp, tier) VALUES
('vm','vm-101','cpu',1.0,?, 'raw'),
@ -411,7 +411,7 @@ func TestStoreQueryDownsampling(t *testing.T) {
}
defer store.Close()
start := time.Unix(1000, 0)
start := recentBucketOffsetTime(300, 100)
for i := 0; i < 10; i++ {
store.writeBatch([]bufferedMetric{
{resourceType: "vm", resourceID: "v1", metricType: "cpu", value: float64(i * 10), timestamp: start.Add(time.Duration(i) * time.Minute), tier: TierRaw},
@ -446,7 +446,7 @@ func TestQueryAllBatch(t *testing.T) {
}
defer store.Close()
ts := time.Unix(1000, 0)
ts := time.Now().UTC().Truncate(time.Second)
store.writeBatch([]bufferedMetric{
// disk-1: smart_temp with 2 points
{resourceType: "disk", resourceID: "disk-1", metricType: "smart_temp", value: 35, timestamp: ts, tier: TierRaw},
@ -577,7 +577,7 @@ func TestQueryAllBatch(t *testing.T) {
}
defer downsampledStore.Close()
base := time.Unix(10_000, 0)
base := time.Now().UTC().Truncate(time.Minute)
downsampledStore.writeBatch([]bufferedMetric{
{resourceType: "disk", resourceID: "disk-ordered", metricType: "smart_temp", value: 33, timestamp: base.Add(130 * time.Second), tier: TierRaw},
{resourceType: "disk", resourceID: "disk-ordered", metricType: "smart_temp", value: 31, timestamp: base.Add(10 * time.Second), tier: TierRaw},
@ -610,7 +610,7 @@ func TestQueryAllBatch(t *testing.T) {
}
defer downsampledStore.Close()
base := time.Unix(20_000, 0)
base := time.Now().UTC().Truncate(time.Minute)
downsampledStore.writeBatch([]bufferedMetric{
{resourceType: "disk", resourceID: "disk-agg", metricType: "smart_temp", value: 10, timestamp: base.Add(5 * time.Second), tier: TierRaw},
{resourceType: "disk", resourceID: "disk-agg", metricType: "smart_temp", value: 16, timestamp: base.Add(25 * time.Second), tier: TierRaw},