reforger_crawler_main/metrics/store.go

package metrics

import (
	"encoding/json"
	"fmt"
	"log"
	"sort"
	"sync"
	"sync/atomic"
	"time"

	"gitea.tbdevent.eu/TBD/reforger_crawler_main/initializers"
	"gitea.tbdevent.eu/TBD/reforger_crawler_main/models"
	"gorm.io/gorm"
	"gorm.io/gorm/clause"
)

const (
	resMinute = 60
	resHour   = 3600
	// points that cannot be written for this long are dropped
	maxRetryAge = 15 * 60
)

var (
	idMu     sync.Mutex
	seriesID = map[*Series]uint{}

	// rolledUntil is the start of the first hour without an hourly rollup.
	rolledUntil atomic.Int64
	// flushMu serializes writers of metric_points (flush, rollup, retention).
	flushMu sync.Mutex
)

func pointToAgg(p *models.MetricPoint) *Agg {
	a := &Agg{Count: p.Count, Sum: p.Sum, Min: p.Min, Max: p.Max, Last: p.Last, LastTs: p.Bucket}
	if p.Hist != "" {
		_ = json.Unmarshal([]byte(p.Hist), &a.Hist)
	}
	return a
}

func aggToPoint(id uint, res int, bucket int64, a *Agg) models.MetricPoint {
	return models.MetricPoint{
		SeriesID: id, Resolution: res, Bucket: bucket,
		Count: a.Count, Sum: a.Sum, Min: a.Min, Max: a.Max, Last: a.Last, Hist: a.histJSON(),
	}
}

// resolveIDs makes sure every series has a metric_series row.
func resolveIDs(list []*Series) (map[*Series]uint, error) {
	out := make(map[*Series]uint, len(list))
	var missing []*Series
	idMu.Lock()
	for _, s := range list {
		if id, ok := seriesID[s]; ok {
			out[s] = id
		} else {
			missing = append(missing, s)
		}
	}
	idMu.Unlock()

	for _, s := range missing {
		bounds := ""
		if len(s.Bounds) > 0 {
			b, _ := json.Marshal(s.Bounds)
			bounds = string(b)
		}
		row := models.MetricSeries{
			Service: s.Service, Instance: s.Instance, Name: s.Name, Labels: s.LabelsJSON,
			Kind: string(s.Kind), Bounds: bounds, Help: s.Help, LastSeen: time.Now().Unix(),
		}
		if err := initializers.DB.Clauses(clause.OnConflict{DoNothing: true}).Create(&row).Error; err != nil {
			return nil, err
		}
		if row.ID == 0 {
			if err := initializers.DB.
				Where("service = ? AND instance = ? AND name = ? AND labels = ?", s.Service, s.Instance, s.Name, s.LabelsJSON).
				First(&row).Error; err != nil {
				return nil, err
			}
		}
		idMu.Lock()
		seriesID[s] = row.ID
		idMu.Unlock()
		out[s] = row.ID
	}
	return out, nil
}

type pointID struct {
	id     uint
	bucket int64
}

// persist merges the given buckets into existing rows and upserts them. The
// read-merge-write runs in one transaction; this process is the only writer.
func persist(tx *gorm.DB, res int, points map[pointID]*Agg) error {
	if len(points) == 0 {
		return nil
	}
	keys := make([]pointID, 0, len(points))
	for k := range points {
		keys = append(keys, k)
	}
	sort.Slice(keys, func(i, j int) bool {
		if keys[i].id != keys[j].id {
			return keys[i].id < keys[j].id
		}
		return keys[i].bucket < keys[j].bucket
	})

	for start := 0; start < len(keys); start += 400 {
		end := min(start+400, len(keys))
		chunk := keys[start:end]
		idSet, bucketSet := map[uint]bool{}, map[int64]bool{}
		for _, k := range chunk {
			idSet[k.id] = true
			bucketSet[k.bucket] = true
		}
		ids := make([]uint, 0, len(idSet))
		for id := range idSet {
			ids = append(ids, id)
		}
		buckets := make([]int64, 0, len(bucketSet))
		for b := range bucketSet {
			buckets = append(buckets, b)
		}

		var existing []models.MetricPoint
		if err := tx.Where("resolution = ? AND series_id IN ? AND bucket IN ?", res, ids, buckets).Find(&existing).Error; err != nil {
			return err
		}
		prev := make(map[pointID]*Agg, len(existing))
		for i := range existing {
			prev[pointID{existing[i].SeriesID, existing[i].Bucket}] = pointToAgg(&existing[i])
		}

		rows := make([]models.MetricPoint, 0, len(chunk))
		for _, k := range chunk {
			a := points[k]
			if old := prev[k]; old != nil {
				old.merge(a)
				a = old
			}
			rows = append(rows, aggToPoint(k.id, res, k.bucket, a))
		}
		err := tx.Clauses(clause.OnConflict{
			Columns:   []clause.Column{{Name: "series_id"}, {Name: "resolution"}, {Name: "bucket"}},
			DoUpdates: clause.AssignmentColumns([]string{"count", "sum", "min", "max", "last", "hist"}),
		}).CreateInBatches(&rows, 200).Error
		if err != nil {
			return err
		}
	}
	return nil
}

type pipelineStatus struct {
	mu            sync.Mutex
	LastFlush     int64  `json:"lastFlush"`
	LastFlushMs   int64  `json:"lastFlushMs"`
	LastFlushErr  string `json:"lastFlushError"`
	LastRollup    int64  `json:"lastRollup"`
	LastRetention int64  `json:"lastRetention"`
	Dropped       int64  `json:"droppedPoints"`
}

var status pipelineStatus

// flush writes every bucket that started before `before`.
func flush(before int64) error {
	flushMu.Lock()
	defer flushMu.Unlock()

	started := time.Now()
	points := defaultAgg.takeClosed(before)
	if len(points) == 0 {
		return nil
	}

	fail := func(err error) error {
		now := time.Now().Unix()
		keep := make(map[pointKey]*Agg)
		var dropped int64
		for k, a := range points {
			if now-k.bucket <= maxRetryAge {
				keep[k] = a
			} else {
				dropped++
			}
		}
		defaultAgg.putBack(keep)
		status.mu.Lock()
		status.LastFlushErr = err.Error()
		status.Dropped += dropped
		status.mu.Unlock()
		if dropped > 0 {
			PointsDropped.Add(float64(dropped))
		}
		return err
	}

	set := map[*Series]bool{}
	for k := range points {
		set[k.s] = true
	}
	list := make([]*Series, 0, len(set))
	for s := range set {
		list = append(list, s)
	}
	ids, err := resolveIDs(list)
	if err != nil {
		return fail(fmt.Errorf("resolve series: %w", err))
	}

	rows := make(map[pointID]*Agg, len(points))
	for k, a := range points {
		rows[pointID{ids[k.s], k.bucket}] = a
	}
	err = initializers.DB.Transaction(func(tx *gorm.DB) error {
		return persist(tx, resMinute, rows)
	})
	if err != nil {
		return fail(fmt.Errorf("write points: %w", err))
	}

	// last_seen lets the catalog hide series that stopped reporting
	idList := make([]uint, 0, len(ids))
	for _, id := range ids {
		idList = append(idList, id)
	}
	for start := 0; start < len(idList); start += 500 {
		end := min(start+500, len(idList))
		initializers.DB.Model(&models.MetricSeries{}).Where("id IN ?", idList[start:end]).
			Update("last_seen", time.Now().Unix())
	}

	status.mu.Lock()
	status.LastFlush = time.Now().Unix()
	status.LastFlushMs = time.Since(started).Milliseconds()
	status.LastFlushErr = ""
	status.mu.Unlock()
	return nil
}

func hourOf(ts int64) int64 { return ts - ts%resHour }

// rollup aggregates complete hours of minute data into hourly points. The last
// rolled hour is redone each time to pick up late telemetry; it is idempotent.
func rollup(now int64) error {
	flushMu.Lock()
	defer flushMu.Unlock()

	// the minute data of an hour is complete ~1 flush after the hour ends
	lastFull := hourOf(now - 2*resMinute)
	start := rolledUntil.Load()
	if start == 0 {
		var maxHour, minMinute *int64
		initializers.DB.Model(&models.MetricPoint{}).Where("resolution = ?", resHour).Select("MAX(bucket)").Scan(&maxHour)
		if maxHour != nil {
			start = *maxHour + resHour
		} else {
			initializers.DB.Model(&models.MetricPoint{}).Where("resolution = ?", resMinute).Select("MIN(bucket)").Scan(&minMinute)
			if minMinute == nil {
				status.mu.Lock()
				status.LastRollup = now
				status.mu.Unlock()
				return nil
			}
			start = hourOf(*minMinute)
		}
	}
	from := start - resHour
	if earliest := hourOf(now - int64(initializers.MetricsMinuteRetentionDays)*86400); from < earliest {
		from = earliest
	}

	for h := from; h < lastFull; h += resHour {
		var rows []models.MetricPoint
		err := initializers.DB.Where("resolution = ? AND bucket >= ? AND bucket < ?", resMinute, h, h+resHour).
			Order("bucket asc").Find(&rows).Error
		if err != nil {
			return err
		}
		if len(rows) == 0 {
			continue
		}
		merged := map[pointID]*Agg{}
		for i := range rows {
			k := pointID{rows[i].SeriesID, h}
			a := merged[k]
			if a == nil {
				a = &Agg{}
				merged[k] = a
			}
			a.merge(pointToAgg(&rows[i]))
		}
		// replace (not merge) so re-rolling an hour is idempotent
		out := make([]models.MetricPoint, 0, len(merged))
		for k, a := range merged {
			out = append(out, aggToPoint(k.id, resHour, k.bucket, a))
		}
		err = initializers.DB.Clauses(clause.OnConflict{
			Columns:   []clause.Column{{Name: "series_id"}, {Name: "resolution"}, {Name: "bucket"}},
			DoUpdates: clause.AssignmentColumns([]string{"count", "sum", "min", "max", "last", "hist"}),
		}).CreateInBatches(&out, 200).Error
		if err != nil {
			return err
		}
	}
	if lastFull > rolledUntil.Load() {
		rolledUntil.Store(lastFull)
	}
	status.mu.Lock()
	status.LastRollup = now
	status.mu.Unlock()
	return nil
}

// retention deletes data older than the configured windows.
func retention(now time.Time) error {
	flushMu.Lock()
	defer flushMu.Unlock()

	day := int64(86400)
	ts := now.Unix()
	steps := []struct {
		what string
		run  func() error
	}{
		{"minute points", func() error {
			return initializers.DB.Where("resolution = ? AND bucket < ?", resMinute, ts-int64(initializers.MetricsMinuteRetentionDays)*day).
				Delete(&models.MetricPoint{}).Error
		}},
		{"hour points", func() error {
			return initializers.DB.Where("resolution = ? AND bucket < ?", resHour, ts-int64(initializers.MetricsHourRetentionDays)*day).
				Delete(&models.MetricPoint{}).Error
		}},
		{"events", func() error {
			return initializers.DB.Where("ts < ?", (ts-int64(initializers.EventsRetentionDays)*day)*1000).
				Delete(&models.ServiceEvent{}).Error
		}},
		{"top items", func() error {
			cutoff := now.UTC().AddDate(0, 0, -initializers.TopItemsRetentionDays).Format("2006-01-02")
			return initializers.DB.Where("day < ?", cutoff).Delete(&models.DailyTopItem{}).Error
		}},
		{"stale series", func() error {
			stale := ts - 30*day
			res := initializers.DB.Where("last_seen < ? AND id NOT IN (SELECT DISTINCT series_id FROM metric_points)", stale).
				Delete(&models.MetricSeries{})
			if res.Error == nil && res.RowsAffected > 0 {
				// forget cached ids so a revived series gets a new row
				idMu.Lock()
				seriesID = map[*Series]uint{}
				idMu.Unlock()
			}
			return res.Error
		}},
	}
	for _, step := range steps {
		if err := step.run(); err != nil {
			log.Printf("metrics retention (%s): %v", step.what, err)
		}
	}
	status.mu.Lock()
	status.LastRetention = now.Unix()
	status.mu.Unlock()
	return nil
}