reaper

package
v6.0.0 Latest Latest
Warning

This package is not in the latest version of its module.

Go to latest
Published: Sep 11, 2026 License: BSD-3-Clause Imports: 33 Imported by: 0

Documentation

Overview

Package reaper is responsible to query the metrics from monitored sources and send measurements to sinks.

Index

Constants

This section is empty.

Variables

This section is empty.

Functions

func CheckFolderExistsAndReadable

func CheckFolderExistsAndReadable(path string) bool

func DoesEmergencyTriggerfileExist

func DoesEmergencyTriggerfileExist(fname string) bool

func GCDSlice

func GCDSlice(vals []int) int

GCDSlice computes GCD across a slice. Returns 0 for empty input.

func GetGoPsutilCPU

func GetGoPsutilCPU(interval time.Duration) (metrics.Measurements, error)

GetGoPsutilCPU simulates "psutil" metric output. Assumes the result from last call as input

func GetGoPsutilDiskPG

func GetGoPsutilDiskPG(pgDirs metrics.Measurements) (metrics.Measurements, error)

func GetGoPsutilDiskTotals

func GetGoPsutilDiskTotals() (metrics.Measurements, error)

func GetGoPsutilMem

func GetGoPsutilMem() (metrics.Measurements, error)

func GetLoadAvgLocal

func GetLoadAvgLocal() (metrics.Measurements, error)

func GetPathUnderlyingDeviceID

func GetPathUnderlyingDeviceID(path string) (uint64, error)

func IsDirectlyFetchableMetric

func IsDirectlyFetchableMetric(md *sources.DbConn, metric string) bool

func QueryMeasurements

func QueryMeasurements(ctx context.Context, md *sources.DbConn, sql string, args ...any) (metrics.Measurements, error)

QueryMeasurements runs the given SQL on the source connection and returns the collected rows. The round-trip is bounded by db.ChangeDetectionTimeout so a hung source cannot block the change-detection sweep indefinitely.

Types

type ChangeDetectionResults

type ChangeDetectionResults struct {
	Target  string
	Created int
	Altered int
	Dropped int
}

func (*ChangeDetectionResults) String

func (cdr *ChangeDetectionResults) String() string

func (*ChangeDetectionResults) Total

func (cdr *ChangeDetectionResults) Total() int

type ConcurrentMetricDefs

type ConcurrentMetricDefs struct {
	*metrics.Metrics
	sync.RWMutex
}

func NewConcurrentMetricDefs

func NewConcurrentMetricDefs() *ConcurrentMetricDefs

func (*ConcurrentMetricDefs) Assign

func (cmd *ConcurrentMetricDefs) Assign(newDefs *metrics.Metrics)

func (*ConcurrentMetricDefs) GetMetricDef

func (cmd *ConcurrentMetricDefs) GetMetricDef(name string) (m metrics.Metric, ok bool)

func (*ConcurrentMetricDefs) GetPresetDef

func (cmd *ConcurrentMetricDefs) GetPresetDef(name string) (m metrics.Preset, ok bool)

func (*ConcurrentMetricDefs) GetPresetMetrics

func (cmd *ConcurrentMetricDefs) GetPresetMetrics(name string) (m metrics.MetricIntervals)

type DbConnReaper

type DbConnReaper struct {
	// contains filtered or unexported fields
}

DbConnReaper manages metric collection for a single monitored database source. Instead of one goroutine per metric it runs a single GCD-based tick loop and batches SQL queries via pgx.Batch when the source is a real Postgres connection (non-pgbouncer, non-pgpool).

func NewDbConnReaper

func NewDbConnReaper(r *reaper, md *sources.DbConn) *DbConnReaper

NewDbConnReaper creates a SourceReaper for the given source connection.

func (*DbConnReaper) CollectAndDispatch

func (sr *DbConnReaper) CollectAndDispatch(ctx context.Context, rows pgx.Rows, name string, metric metrics.Metric) error

CollectAndDispatch is a helper that collects rows from a pgx.Rows and dispatches them.

func (*DbConnReaper) Reap

func (sr *DbConnReaper) Reap(ctx context.Context)

Run is the main loop for a single source. It replaces N per-metric goroutines with one goroutine that batches SQL queries at GCD-aligned ticks.

type ExistingPartitionInfo

type ExistingPartitionInfo struct {
	StartTime time.Time
	EndTime   time.Time
}

type InstanceMetricCache

type InstanceMetricCache struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

func NewInstanceMetricCache

func NewInstanceMetricCache() *InstanceMetricCache

func (*InstanceMetricCache) Get

func (*InstanceMetricCache) Put

func (imc *InstanceMetricCache) Put(key string, data metrics.Measurements)

type LogConfig

type LogConfig struct {
	CollectorEnabled   bool
	CSVDestination     bool
	TruncateOnRotation bool
	Directory          string
	ServerMessagesLang string
}

type LogParser

type LogParser struct {
	*LogConfig

	LogsMatchRegex *regexp.Regexp
	SourceConn     *sources.DbConn

	Interval time.Duration
	StoreCh  chan<- metrics.MeasurementEnvelope
	// contains filtered or unexported fields
}

func NewLogParser

func NewLogParser(ctx context.Context, mdb *sources.DbConn, storeCh chan<- metrics.MeasurementEnvelope) (lp *LogParser, err error)

func (*LogParser) GetMeasurementEnvelope

func (lp *LogParser) GetMeasurementEnvelope() metrics.MeasurementEnvelope

GetMeasurementEnvelope converts current event counts to a MeasurementEnvelope

func (*LogParser) HasSendIntervalElapsed

func (lp *LogParser) HasSendIntervalElapsed() bool

func (*LogParser) ParseLogs

func (lp *LogParser) ParseLogs() error

type PromReaper

type PromReaper struct {
	// contains filtered or unexported fields
}

PromReaper drives metric scraping for a single Prometheus source. It runs a GCD-based tick loop and applies per-family emit-interval gating.

func NewPromSourceReaper

func NewPromSourceReaper(r *reaper, md *sources.PromConn) *PromReaper

NewPromSourceReaper creates a PromReaper for the given Prometheus source.

func (*PromReaper) Reap

func (pr *PromReaper) Reap(ctx context.Context)

Reap is the main loop for a Prometheus source. It scrapes all metric families on every GCD tick and emits envelopes that have passed their per-family interval.

func (*PromReaper) ScrapeAll

func (pr *PromReaper) ScrapeAll(ctx context.Context) ([]metrics.MeasurementEnvelope, error)

ScrapeAll fetches Prometheus exposition metrics from pr.md and returns one MeasurementEnvelope per metric family. Each sample becomes one Measurement with tag_<label> columns (skipping __name__), a value column named after the family, and epoch_ns set from the sample timestamp (ms→ns) or time.Now().

type Readier

type Readier interface {
	Ready() bool
}

type ReadierReaper

type ReadierReaper interface {
	Reaper
	Readier
}

func NewReaper

func NewReaper(ctx context.Context, opts *cmdopts.Options) ReadierReaper

type Reaper

type Reaper interface {
	Reap(ctx context.Context)
}

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL