stats

Documentation

Overview

Package stats provides the store for measurements associated with agents and configurations.

Index

• Constants
• Variables
• func Agent(m *record.Metric) string
• func Configuration(m *record.Metric) string
• func GetDurationFromPeriod(opts QueryOptions) time.Duration
• func IsNewMeasurementsFormat(m *record.Metric) bool
• func ParseProcessorName(processorName string) (position string, pipelineType string, name string)
• func Processor(m *record.Metric) string
• func ProcessorParsed(m *record.Metric) (position string, pipelineType string, name string)
• func Value(m *record.Metric) (value float64, ok bool)
• type DefaultBatcher
  • func NewDefaultBatcher(ctx context.Context, logger *zap.Logger, measurements Measurements) *DefaultBatcher
  • func (d *DefaultBatcher) AcceptMetrics(ctx context.Context, metrics []*record.Metric) error
  • func (d *DefaultBatcher) Shutdown(ctx context.Context) error
• type MeasurementBatcher
• type Measurements
• type MetricData
• type QueryOption
  • func WithPeriod(period time.Duration) QueryOption
• type QueryOptions
  • func MakeQueryOptions(options []QueryOption) QueryOptions
• type RateMetric
  • func Rate(first, last RateMetric) (*RateMetric, error)

Constants

const (
	AgentAttributeName         = "agent"
	ConfigurationAttributeName = "configuration"
	ProcessorAttributeName     = "processor"
	LogDataSizeMetricName      = "otelcol_processor_throughputmeasurement_log_data_size"
	MetricDataSizeMetricName   = "otelcol_processor_throughputmeasurement_metric_data_size"
	TraceDataSizeMetricName    = "otelcol_processor_throughputmeasurement_trace_data_size"
)

Metric Attribute names used for measurements and Metric Names as they come exported with prometheus

Variables

var MeasurementBatchFlushInterval = 5 * time.Second

MeasurementBatchFlushInterval is the number of seconds to wait before flushing batch.

var SupportedMetricNames = []string{
	LogDataSizeMetricName,
	MetricDataSizeMetricName,
	TraceDataSizeMetricName,
}

SupportedMetricNames is the list of metrics we care about coming from the self-monitoring of the agent

Functions

func Agent

func Agent(m *record.Metric) string

Agent returns the value of the metric attribute that corresponds to the processor name or "" if the attribute is missing or not a string.

func Configuration

func Configuration(m *record.Metric) string

Configuration returns the value of the metric attribute that corresponds to the configuration name or "" if the attribute is missing or not a string.

func GetDurationFromPeriod

func GetDurationFromPeriod(opts QueryOptions) time.Duration

GetDurationFromPeriod returns the rollup duration for a given query period

func IsNewMeasurementsFormat

func IsNewMeasurementsFormat(m *record.Metric) bool

IsNewMeasurementsFormat returns true if the metric is in the new format, false if it is in the old format

If the processor name is blank or missing, we return true If the processor name indicates it is for a destination, we return true

TODO: Remove this function once we are sure all configurations are using the new format

func ParseProcessorName

func ParseProcessorName(processorName string) (position string, pipelineType string, name string)

ParseProcessorName parses a processor name like throughputmeasurement/_s0_metrics_source0, returning the position, type, and name of the processor.

func Processor

func Processor(m *record.Metric) string

Processor returns the value of the metric attribute that corresponds to the processor name or "" if the attribute is missing or not a string.

func ProcessorParsed

func ProcessorParsed(m *record.Metric) (position string, pipelineType string, name string)

ProcessorParsed returns the individual parts of the processor name parsed out of the processor attribute

func Value

func Value(m *record.Metric) (value float64, ok bool)

Value returns the value of the metric as a float64

Types

type DefaultBatcher

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

DefaultBatcher is the default implementation of the MeasurementBatcher

func NewDefaultBatcher

func NewDefaultBatcher(ctx context.Context, logger *zap.Logger, measurements Measurements) *DefaultBatcher

NewDefaultBatcher creates a new default MeasurementBatcher

func (*DefaultBatcher) AcceptMetrics

func (d *DefaultBatcher) AcceptMetrics(ctx context.Context, metrics []*record.Metric) error

AcceptMetrics accepts metrics to be batched

func (*DefaultBatcher) Shutdown

func (d *DefaultBatcher) Shutdown(ctx context.Context) error

Shutdown shuts down the batcher

type MeasurementBatcher

type MeasurementBatcher interface {
	// AcceptMetrics adds the metrics to the batcher that will eventually be saved in storage
	AcceptMetrics(ctx context.Context, metrics []*record.Metric) error

	// Shutdown stops the batcher
	Shutdown(ctx context.Context) error
}

MeasurementBatcher the metric batcher accepts metrics and batches them before saving them in storage

type Measurements

type Measurements interface {
	// Clear clears the store and is mostly used for testing.
	Clear()

	// MeasurementsSize returns the count of keys in the store, and is used only for testing
	MeasurementsSize(context.Context) (int, error)

	// AgentMetrics provides metrics for an individual agents. They are essentially configuration metrics filtered to a
	// list of agents.
	//
	// Note: While the same record.Metric struct is used to return the metrics, these are not the same metrics provided to
	// Store. They will be aggregated and counter metrics will be converted into rates.
	AgentMetrics(ctx context.Context, id []string, options ...QueryOption) (MetricData, error)

	// ConfigurationMetrics provides all metrics associated with a configuration aggregated from all agents using the
	// configuration.
	//
	// Note: While the same record.Metric struct is used to return the metrics, these are not the same metrics provided to
	// Store. They will be aggregated and counter metrics will be converted into rates.
	ConfigurationMetrics(ctx context.Context, name string, options ...QueryOption) (MetricData, error)

	// OverviewMetrics provides all metrics needed for the overview page. This page shows configurations and destinations.
	// The metrics required are the MeasurementPositionDestinationAfterProcessors metric of each configuration which has a
	// processor with the prefix "throughputmeasurement/_d1_"
	OverviewMetrics(ctx context.Context, options ...QueryOption) (MetricData, error)

	// SaveAgentMetrics saves new metrics. These metrics will be aggregated to determine metrics associated with agents and configurations.
	SaveAgentMetrics(ctx context.Context, metrics []*record.Metric) error

	// ProcessMetrics is called in the background at regular intervals and performs metric roll-up and removes old data
	ProcessMetrics(ctx context.Context) error
}

Measurements provides query and storage of time-series metrics associated with agents and configurations.

type MetricData

type MetricData []*record.Metric

MetricData is returned by Measurements when metrics are requested for agents and configurations

type QueryOption

type QueryOption func(*QueryOptions)

QueryOption is an option used in Store queries

func WithPeriod

func WithPeriod(period time.Duration) QueryOption

WithPeriod specifies the period for which the metrics should be returned

type QueryOptions

type QueryOptions struct {
	Period time.Duration
}

QueryOptions represents the set of options available for a measurements query

func MakeQueryOptions

func MakeQueryOptions(options []QueryOption) QueryOptions

MakeQueryOptions constructs a QueryOptions struct from the requested options

type RateMetric

type RateMetric struct {
	Timestamp      time.Time
	StartTimestamp time.Time
	Value          float64
}

RateMetric is a metric that can be used to calculate a rate of change, as the Rate function can be called from locations using pmetrics or pgMeasurements

func Rate

func Rate(first, last RateMetric) (*RateMetric, error)

Rate returns the rate of change of the metric over the time period between the two metrics