reforger_crawler_main/metrics/pipeline.go

package metrics

import (
	"context"
	"log"
	"math"
	"sync"
	"time"

	"gitea.tbdevent.eu/TBD/reforger_crawler_main/initializers"
)

var startOnce sync.Once

// Start launches the background jobs: minute flush and samplers, top-N
// persistence, hourly rollup and retention, and service heartbeat checks.
func Start(ctx context.Context) {
	startOnce.Do(func() {
		Top.seed()
		go every(ctx, "flush", time.Minute, 5*time.Second, func(now time.Time) {
			sample()
			if err := flush(minuteOf(now.Unix())); err != nil {
				log.Printf("metrics: flush failed: %v", err)
			}
		})
		go every(ctx, "top", 5*time.Minute, 0, func(time.Time) {
			if err := Top.flush(); err != nil {
				log.Printf("metrics: top items flush failed: %v", err)
			}
		})
		go every(ctx, "rollup", time.Hour, 2*time.Minute, func(now time.Time) {
			RunMaintenance(now)
		})
		go every(ctx, "services", 30*time.Second, 0, func(now time.Time) {
			selfHeartbeat(now)
			checkServices(now)
		})
		// initialize the rollup watermark and service state right away
		go func() {
			selfHeartbeat(time.Now())
			checkServices(time.Now())
			if err := rollup(time.Now().Unix()); err != nil {
				log.Printf("metrics: rollup failed: %v", err)
			}
		}()
	})
}

// RunMaintenance runs the rollup and retention jobs.
func RunMaintenance(now time.Time) {
	if err := rollup(now.Unix()); err != nil {
		log.Printf("metrics: rollup failed: %v", err)
	}
	if err := retention(now); err != nil {
		log.Printf("metrics: retention failed: %v", err)
	}
}

// every runs fn on a fixed period aligned to wall clock (+offset).
func every(ctx context.Context, name string, period, offset time.Duration, fn func(time.Time)) {
	for {
		now := time.Now()
		next := now.Truncate(period).Add(offset)
		if !next.After(now) {
			next = next.Add(period)
		}
		select {
		case <-ctx.Done():
			return
		case t := <-time.After(time.Until(next)):
			func() {
				defer func() {
					if r := recover(); r != nil {
						log.Printf("metrics: %s job panicked: %v", name, r)
					}
				}()
				fn(t)
			}()
		}
	}
}

// FlushAll writes everything still in memory; used on shutdown.
func FlushAll() {
	sample()
	if err := flush(math.MaxInt64); err != nil {
		log.Printf("metrics: final flush failed: %v", err)
	}
	if err := Top.flush(); err != nil {
		log.Printf("metrics: final top items flush failed: %v", err)
	}
}

// QueueCounts is a cheap snapshot of the indexing queue.
type QueueCounts struct {
	Total    int64 `json:"total"`
	Queued   int64 `json:"queued"`
	Priority int64 `json:"priority"`
	Indexing int64 `json:"indexing"`
}

// LoadQueueCounts counts addons by queue state with one query.
func LoadQueueCounts() (QueueCounts, error) {
	var qc QueueCounts
	err := initializers.DB.Raw(`SELECT COUNT(*) AS total,
		COALESCE(SUM(CASE WHEN to_be_indexed = ? AND is_being_indexed = ? THEN 1 ELSE 0 END), 0) AS queued,
		COALESCE(SUM(CASE WHEN to_be_indexed = ? AND is_being_indexed = ? AND priority_indexing = ? THEN 1 ELSE 0 END), 0) AS priority,
		COALESCE(SUM(CASE WHEN is_being_indexed = ? THEN 1 ELSE 0 END), 0) AS indexing
		FROM addons WHERE deleted_at IS NULL`,
		true, false, true, false, true, true).Scan(&qc).Error
	return qc, err
}

// sample records gauges that are read from the database or memory.
func sample() {
	if qc, err := LoadQueueCounts(); err == nil {
		AddonsTotal.Set(float64(qc.Total))
		QueueDepth.Set(float64(qc.Queued), "total")
		QueueDepth.Set(float64(qc.Priority), "priority")
		BeingIndexed.Set(float64(qc.Indexing))
	}
	humans, bots := liveVisitors()
	VisitorsUnique.Set(float64(humans), "human")
	VisitorsUnique.Set(float64(bots), "bot")
}

// PipelineStatus describes the health of the metrics pipeline.
type PipelineStatus struct {
	Enabled       bool   `json:"enabled"`
	LastFlush     int64  `json:"lastFlush"`
	LastFlushMs   int64  `json:"lastFlushMs"`
	LastFlushErr  string `json:"lastFlushError"`
	LastRollup    int64  `json:"lastRollup"`
	LastRetention int64  `json:"lastRetention"`
	RolledUntil   int64  `json:"rolledUntil"`
	OpenPoints    int    `json:"openPoints"`
	LiveSeries    int    `json:"liveSeries"`
	DroppedPoints int64  `json:"droppedPoints"`
}

func Status() PipelineStatus {
	status.mu.Lock()
	defer status.mu.Unlock()
	return PipelineStatus{
		Enabled:       initializers.MetricsEnabled,
		LastFlush:     status.LastFlush,
		LastFlushMs:   status.LastFlushMs,
		LastFlushErr:  status.LastFlushErr,
		LastRollup:    status.LastRollup,
		LastRetention: status.LastRetention,
		RolledUntil:   rolledUntil.Load(),
		OpenPoints:    defaultAgg.openCount(),
		LiveSeries:    len(reg.all()),
		DroppedPoints: status.Dropped,
	}
}