stages

package
v0.40.4 Latest Latest
Warning

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

 Go to latest Published: Apr 12, 2024 License: Apache-2.0 Imports: 48 Imported by: 0

Documentation

Index

Constants

View Source 
const (
	ErrDropStageEmptyConfig       = "drop stage config must contain at least one of `source`, `expression`, `older_than` or `longer_than`"
	ErrDropStageInvalidConfig     = "drop stage config error, `value` and `expression` cannot both be defined at the same time."
	ErrDropStageInvalidRegex      = "drop stage regex compilation error: %v"
	ErrDropStageNoSourceWithValue = "drop stage config must contain `source` if `value` is specified"
)
View Source 
const (
	ErrExpressionsRequired  = "JMES expression is required"
	ErrCouldNotCompileJMES  = "could not compile JMES expression"
	ErrEmptyJSONStageConfig = "empty json stage configuration"
	ErrEmptyJSONStageSource = "empty source"
	ErrMalformedJSON        = "malformed json"
)

Config Errors

View Source 
const (
	ErrEmptyLabelStageConfig = "label stage config cannot be empty"
	ErrInvalidLabelName      = "invalid label name: %s"
)
View Source 
const (
	MetricTypeCounter   = "counter"
	MetricTypeGauge     = "gauge"
	MetricTypeHistogram = "histogram"
)

Metric types.

View Source 
const (
	StageTypeCRI        = "cri"
	StageTypeDecolorize = "decolorize"
	StageTypeDocker     = "docker"
	StageTypeDrop       = "drop"
	//TODO(thampiotr): Add support for eventlogmessage stage
	StageTypeEventLogMessage    = "eventlogmessage"
	StageTypeGeoIP              = "geoip"
	StageTypeJSON               = "json"
	StageTypeLabel              = "labels"
	StageTypeLabelAllow         = "labelallow"
	StageTypeLabelDrop          = "labeldrop"
	StageTypeLimit              = "limit"
	StageTypeLogfmt             = "logfmt"
	StageTypeMatch              = "match"
	StageTypeMetric             = "metrics"
	StageTypeMultiline          = "multiline"
	StageTypeOutput             = "output"
	StageTypePack               = "pack"
	StageTypePipeline           = "pipeline"
	StageTypeRegex              = "regex"
	StageTypeReplace            = "replace"
	StageTypeSampling           = "sampling"
	StageTypeStaticLabels       = "static_labels"
	StageTypeStructuredMetadata = "structured_metadata"
	StageTypeTemplate           = "template"
	StageTypeTenant             = "tenant"
	StageTypeTimestamp          = "timestamp"
)

TODO(@tpaschalis) Let's use this as the list of stages we need to port over.

View Source 
const (
	ErrSamplingStageInvalidRate = "sampling stage failed to parse rate,Sampling Rate must be between 0.0 and 1.0, received %f"
)
View Source 
const (
	ErrTimestampContainsYear = "timestamp '%s' is expected to not contain the year date component"
)
View Source 
const MinReasonableMaxDistinctLabels = 10000 // 80bytes per rate.Limiter ~ 1MiB memory

MinReasonableMaxDistinctLabels provides a sensible default.

View Source 
const (
	RFC3339Nano = "RFC3339Nano"
)
View Source 
const ReservedLabelTenantID = "__tenant_id__"

ReservedLabelTenantID is a shared value used to refer to the tenant ID.

Variables

View Source 
var (
	ErrEmptyGeoIPStageConfig                = errors.New("geoip stage config cannot be empty")
	ErrEmptyDBPathGeoIPStageConfig          = errors.New("db path cannot be empty")
	ErrEmptySourceGeoIPStageConfig          = errors.New("source cannot be empty")
	ErrEmptyDBTypeGeoIPStageConfig          = errors.New("db type should be either city or asn")
	ErrEmptyDBTypeAndValuesGeoIPStageConfig = errors.New("db type or values need to be set")
)
View Source 
var (
	ErrLimitStageInvalidRateOrBurst = errors.New("limit stage failed to parse rate or burst")
	ErrLimitStageByLabelMustDrop    = errors.New("When ratelimiting by label, drop must be true")
)

Configuration errors.

View Source 
var (
	ErrMappingRequired        = errors.New("logfmt mapping is required")
	ErrEmptyLogfmtStageConfig = errors.New("empty logfmt stage configuration")
)

Config Errors

View Source 
var (
	ErrEmptyMatchStageConfig = errors.New("match stage config cannot be empty")
	ErrPipelineNameRequired  = errors.New("match stage pipeline name can be omitted but cannot be an empty string")
	ErrSelectorRequired      = errors.New("selector statement required for match stage")
	ErrMatchRequiresStages   = errors.New("match stage requires at least one additional stage to be defined in '- stages'")
	ErrSelectorSyntax        = errors.New("invalid selector syntax for match stage")
	ErrStagesWithDropLine    = errors.New("match stage configured to drop entries cannot contains stages")
	ErrUnknownMatchAction    = errors.New("match stage action should be 'keep' or 'drop'")

	MatchActionKeep = "keep"
	MatchActionDrop = "drop"
)

Configuration errors.

View Source 
var (
	ErrEmptyMetricsStageConfig = errors.New("empty metric stage configuration")
	ErrMetricsStageInvalidType = errors.New("invalid metric type: must be one of 'counter', 'gauge', or 'histogram'")
	ErrInvalidIdleDur          = errors.New("max_idle_duration could not be parsed as a time.Duration")
	ErrSubSecIdleDur           = errors.New("max_idle_duration less than 1s not allowed")
)

Configuration errors.

View Source 
var (
	ErrMultilineStageEmptyConfig        = errors.New("multiline stage config must define `firstline` regular expression")
	ErrMultilineStageInvalidRegex       = errors.New("multiline stage first line regex compilation error")
	ErrMultilineStageInvalidMaxWaitTime = errors.New("multiline stage `max_wait_time` parse error")
)

Configuration errors.

View Source 
var (
	ErrEmptyOutputStageConfig = errors.New("output stage config cannot be empty")
	ErrOutputSourceRequired   = errors.New("output source value is required if output is specified")
)

Config Errors.

View Source 
var (
	ErrExpressionRequired    = errors.New("expression is required")
	ErrCouldNotCompileRegex  = errors.New("could not compile regular expression")
	ErrEmptyRegexStageSource = errors.New("empty source")
)

Config Errors.

View Source 
var (
	ErrEmptyReplaceStageConfig = errors.New("empty replace stage configuration")
	ErrEmptyReplaceStageSource = errors.New("empty source in replace stage")
)

Config Errors

View Source 
var (
	ErrEmptyTemplateStageConfig = errors.New("template stage config cannot be empty")
	ErrTemplateSourceRequired   = errors.New("template source value is required")
)

Config Errors.

View Source 
var (
	ErrTenantStageEmptyLabelSourceOrValue        = errors.New("label, source or value config are required")
	ErrTenantStageConflictingLabelSourceAndValue = errors.New("label, source and value are mutually exclusive: you should set source, value or label but not all")
)

Configuration errors.

View Source 
var (
	ErrEmptyTimestampStageConfig = errors.New("timestamp stage config cannot be empty")
	ErrTimestampSourceRequired   = errors.New("timestamp source value is required if timestamp is specified")
	ErrTimestampFormatRequired   = errors.New("timestamp format is required")
	ErrInvalidLocation           = errors.New("invalid location specified: %v")
	ErrInvalidActionOnFailure    = errors.New("invalid action on failure (supported values are %v)")
	ErrTimestampSourceMissing    = errors.New("extracted data did not contain a timestamp")
	ErrTimestampConversionFailed = errors.New("failed to convert extracted time to string")
	ErrTimestampParsingFailed    = errors.New("failed to parse time")

	Unix   = "Unix"
	UnixMs = "UnixMs"
	UnixUs = "UnixUs"
	UnixNs = "UnixNs"

	TimestampActionOnFailureSkip    = "skip"
	TimestampActionOnFailureFudge   = "fudge"
	TimestampActionOnFailureDefault = TimestampActionOnFailureFudge
)

Config errors.

View Source 
var (
	// Debug is used to wrap debug log statements, the go-kit logger won't let us introspect the current log level
	// so this global is used for that purpose. This allows us to skip allocations of log messages at the
	// debug level when debug level logging is not enabled. Log level allocations can become very expensive
	// as we log numerous log entries per log line at debug level.
	Debug = false

	// Inspect is used to debug promtail pipelines by showing diffs between pipeline stages
	Inspect = false
)
View Source 
var DefaultCRIConfig = CRIConfig{
	MaxPartialLines:            100,
	MaxPartialLineSize:         0,
	MaxPartialLineSizeTruncate: false,
}

DefaultCRIConfig contains the default CRIConfig values.

View Source 
var DefaultMultilineConfig = MultilineConfig{
	MaxLines:    128,
	MaxWaitTime: 3 * time.Second,
}

DefaultMultilineConfig applies the default values on

View Source 
var DefaultPackConfig = PackConfig{
	IngestTimestamp: true,
}

DefaultPackConfig sets the defaults.

View Source 
var ErrEmptyLabelAllowStageConfig = errors.New("labelallow stage config cannot be empty")

ErrEmptyLabelAllowStageConfig error is returned if the config is empty.

View Source 
var ErrEmptyLabelDropStageConfig = errors.New("labeldrop stage config cannot be empty")

ErrEmptyLabelDropStageConfig error returned if the config is empty.

View Source 
var ErrEmptyStaticLabelStageConfig = errors.New("static_labels stage config cannot be empty")

ErrEmptyStaticLabelStageConfig error returned if the config is empty.

View Source 
var TimestampActionOnFailureOptions = []string{TimestampActionOnFailureSkip, TimestampActionOnFailureFudge}

TimestampActionOnFailureOptions defines the available options for the `action_on_failure` field.

Functions

func RunWith 

func RunWith(input chan Entry, process func(e Entry) Entry) chan Entry

RunWith will reads from the input channel entries, mutate them with the process function and returns them via the output channel.

func RunWithSkip 

func RunWithSkip(input chan Entry, process func(e Entry) (Entry, bool)) chan Entry

RunWithSkip same as RunWith, except it skip sending it to output channel, if `process` functions returns `skip` true.

func RunWithSkipOrSendMany 

func RunWithSkipOrSendMany(input chan Entry, process func(e Entry) ([]Entry, bool)) chan Entry

RunWithSkiporSendMany same as RunWithSkip, except it can either skip sending it to output channel, if `process` functions returns `skip` true. Or send many entries.

func SanitizeFullLabelName added in v0.38.0 

func SanitizeFullLabelName(input string) string

Sanitize a input string to convert it into a valid prometheus label TODO: switch to prometheus/prometheus/util/strutil/SanitizeFullLabelName

func SetReadLineRateLimiter 

func SetReadLineRateLimiter(rateVal float64, burstVal int, drop bool)

Types

type CRIConfig 

type CRIConfig struct {
	MaxPartialLines            int    `river:"max_partial_lines,attr,optional"`
	MaxPartialLineSize         uint64 `river:"max_partial_line_size,attr,optional"`
	MaxPartialLineSizeTruncate bool   `river:"max_partial_line_size_truncate,attr,optional"`
}

CRIConfig is an empty struct that is used to enable a pre-defined pipeline for decoding entries that are using the CRI logging format.

func (*CRIConfig) SetToDefault added in v0.37.0 

func (args *CRIConfig) SetToDefault()

SetToDefault implements river.Defaulter.

func (*CRIConfig) Validate added in v0.37.0 

func (args *CRIConfig) Validate() error

Validate implements river.Validator.

type DecolorizeConfig added in v0.38.0 

type DecolorizeConfig struct{}

type DockerConfig 

type DockerConfig struct{}

DockerConfig is an empty struct that is used to enable a pre-defined pipeline for decoding entries that are using the Docker logs format.

type DropConfig 

type DropConfig struct {
	DropReason string           `river:"drop_counter_reason,attr,optional"`
	Source     string           `river:"source,attr,optional"`
	Value      string           `river:"value,attr,optional"`
	Separator  string           `river:"separator,attr,optional"`
	Expression string           `river:"expression,attr,optional"`
	OlderThan  time.Duration    `river:"older_than,attr,optional"`
	LongerThan units.Base2Bytes `river:"longer_than,attr,optional"`
	// contains filtered or unexported fields
}

DropConfig contains the configuration for a dropStage

type Entry 

type Entry struct {
	Extracted map[string]interface{}
	loki.Entry
}

type EventLogMessageConfig added in v0.38.0 

type EventLogMessageConfig struct {
	Source            string `river:"source,attr,optional"`
	DropInvalidLabels bool   `river:"drop_invalid_labels,attr,optional"`
	OverwriteExisting bool   `river:"overwrite_existing,attr,optional"`
}

func (*EventLogMessageConfig) SetToDefault added in v0.38.0 

func (e *EventLogMessageConfig) SetToDefault()

func (*EventLogMessageConfig) Validate added in v0.38.0 

func (e *EventLogMessageConfig) Validate() error

type GenerationalMap 

type GenerationalMap[K comparable, V any] struct {
	// contains filtered or unexported fields
}

GenerationalMap is ported from Loki's pkg/util package. It didn't exist in our dependency at the time, so I copied the implementation over.

func NewGenMap 

func NewGenMap[K comparable, V any](maxSize int, newV func() V, gcCb func()) GenerationalMap[K, V]

NewGenMap created which maintains at most maxSize recently used entries

func (*GenerationalMap[K, T]) GetOrCreate 

func (m *GenerationalMap[K, T]) GetOrCreate(key K) T

type GeoIPConfig 

type GeoIPConfig struct {
	DB            string            `river:"db,attr"`
	Source        *string           `river:"source,attr"`
	DBType        string            `river:"db_type,attr,optional"`
	CustomLookups map[string]string `river:"custom_lookups,attr,optional"`
}

GeoIPConfig represents GeoIP stage config

type GeoIPFields 

type GeoIPFields int
const (
	CITYNAME GeoIPFields = iota
	COUNTRYNAME
	COUNTRYCODE
	CONTINENTNAME
	CONTINENTCODE
	LOCATION
	POSTALCODE
	TIMEZONE
	SUBDIVISIONNAME
	SUBDIVISIONCODE
	ASN
	ASNORG
)

type JSONConfig 

type JSONConfig struct {
	Expressions   map[string]string `river:"expressions,attr"`
	Source        *string           `river:"source,attr,optional"`
	DropMalformed bool              `river:"drop_malformed,attr,optional"`
}

JSONConfig represents a JSON Stage configuration

type LabelAllowConfig 

type LabelAllowConfig struct {
	Values []string `river:"values,attr"`
}

LabelAllowConfig contains the slice of labels to allow through.

type LabelDropConfig 

type LabelDropConfig struct {
	Values []string `river:"values,attr"`
}

LabelDropConfig contains the slice of labels to be dropped.

type LabelsConfig 

type LabelsConfig struct {
	Values map[string]*string `river:"values,attr"`
}

LabelsConfig is a set of labels to be extracted

type LimitConfig 

type LimitConfig struct {
	Rate              float64 `river:"rate,attr"`
	Burst             int     `river:"burst,attr"`
	Drop              bool    `river:"drop,attr,optional"`
	ByLabelName       string  `river:"by_label_name,attr,optional"`
	MaxDistinctLabels int     `river:"max_distinct_labels,attr,optional"`
}

LimitConfig sets up a Limit stage.

type LogfmtConfig 

type LogfmtConfig struct {
	Mapping map[string]string `river:"mapping,attr"`
	Source  string            `river:"source,attr,optional"`
}

LogfmtConfig represents a logfmt Stage configuration

type MatchConfig 

type MatchConfig struct {
	Selector     string        `river:"selector,attr"`
	Stages       []StageConfig `river:"stage,enum,optional"`
	Action       string        `river:"action,attr,optional"`
	PipelineName string        `river:"pipeline_name,attr,optional"`
	DropReason   string        `river:"drop_counter_reason,attr,optional"`
}

MatchConfig contains the configuration for a matcherStage

type MetricConfig 

type MetricConfig struct {
	Counter   *metric.CounterConfig   `river:"counter,block,optional"`
	Gauge     *metric.GaugeConfig     `river:"gauge,block,optional"`
	Histogram *metric.HistogramConfig `river:"histogram,block,optional"`
}

MetricConfig is a single metrics configuration. TODO(@tpaschalis) Rework once River squashing is implemented.

type MetricsConfig 

type MetricsConfig struct {
	Metrics []MetricConfig `river:"metric,enum,optional"`
}

MetricsConfig is a set of configured metrics.

type MultilineConfig 

type MultilineConfig struct {
	Expression  string        `river:"firstline,attr"`
	MaxLines    uint64        `river:"max_lines,attr,optional"`
	MaxWaitTime time.Duration `river:"max_wait_time,attr,optional"`
	// contains filtered or unexported fields
}

MultilineConfig contains the configuration for a Multiline stage.

func (*MultilineConfig) SetToDefault 

func (args *MultilineConfig) SetToDefault()

SetToDefault implements river.Defaulter.

func (*MultilineConfig) Validate 

func (args *MultilineConfig) Validate() error

Validate implements river.Validator.

type OutputConfig 

type OutputConfig struct {
	Source string `river:"source,attr"`
}

OutputConfig initializes a configuration stage which sets the log line to a value from the extracted map.

type PackConfig 

type PackConfig struct {
	Labels          []string `river:"labels,attr"`
	IngestTimestamp bool     `river:"ingest_timestamp,attr,optional"`
}

PackConfig contains the configuration for a packStage

func (*PackConfig) SetToDefault 

func (p *PackConfig) SetToDefault()

SetToDefault implements river.Defaulter.

type Packed 

type Packed struct {
	Labels map[string]string `json:",inline"`
	Entry  string            `json:"_entry"`
}

Packed keeps track of the labels and log entry.

func (Packed) MarshalJSON 

func (w Packed) MarshalJSON() ([]byte, error)

MarshalJSON creates a Packed struct as JSON where the Labels are flattened into the top level of the object

func (*Packed) UnmarshalJSON 

func (w *Packed) UnmarshalJSON(data []byte) error

UnmarshalJSON populates a Packed struct where every key except the _entry key is added to the Labels field

type Pipeline 

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

Pipeline pass down a log entry to each stage for mutation and/or label extraction.

func NewPipeline 

func NewPipeline(logger log.Logger, stages []StageConfig, jobName *string, registerer prometheus.Registerer) (*Pipeline, error)

NewPipeline creates a new log entry pipeline from a configuration

func (*Pipeline) Name 

func (p *Pipeline) Name() string

Name implements Stage

func (*Pipeline) Run 

func (p *Pipeline) Run(in chan Entry) chan Entry

Run implements Stage

func (*Pipeline) Size 

func (p *Pipeline) Size() int

Size gets the current number of stages in the pipeline

func (*Pipeline) Wrap 

func (p *Pipeline) Wrap(next loki.EntryHandler) loki.EntryHandler

Wrap implements EntryMiddleware

type Processor 

type Processor interface {
	Process(labels model.LabelSet, extracted map[string]interface{}, time *time.Time, entry *string)
	Name() string
}

Processor takes an existing set of labels, timestamp and log entry and returns either a possibly mutated timestamp and log entry

type RegexConfig 

type RegexConfig struct {
	Expression string  `river:"expression,attr"`
	Source     *string `river:"source,attr,optional"`
}

RegexConfig configures a processing stage uses regular expressions to extract values from log lines into the shared values map.

type ReplaceConfig 

type ReplaceConfig struct {
	Expression string `river:"expression,attr"`
	Source     string `river:"source,attr,optional"`
	Replace    string `river:"replace,attr,optional"`
}

ReplaceConfig contains a regexStage configuration

type SamplingConfig added in v0.38.0 

type SamplingConfig struct {
	DropReason   *string `river:"drop_counter_reason,attr,optional"`
	SamplingRate float64 `river:"rate,attr"`
}

SamplingConfig contains the configuration for a samplingStage

func (*SamplingConfig) SetToDefault added in v0.38.0 

func (s *SamplingConfig) SetToDefault()

func (*SamplingConfig) Validate added in v0.38.0 

func (s *SamplingConfig) Validate() error

type Stage 

type Stage interface {
	Name() string
	Run(chan Entry) chan Entry
}

Stage can receive entries via an inbound channel and forward mutated entries to an outbound channel.

func New 

func New(logger log.Logger, jobName *string, cfg StageConfig, registerer prometheus.Registerer) (Stage, error)

New creates a new stage for the given type and configuration.

func NewCRI 

func NewCRI(logger log.Logger, config CRIConfig, registerer prometheus.Registerer) (Stage, error)

NewCRI creates a predefined pipeline for parsing entries in the CRI log format.

func NewDocker 

func NewDocker(logger log.Logger, registerer prometheus.Registerer) (Stage, error)

NewDocker creates a predefined pipeline for parsing entries in the Docker json log format.

type StageConfig 

type StageConfig struct {
	//TODO(thampiotr): sync these with new stages
	CRIConfig             *CRIConfig             `river:"cri,block,optional"`
	DecolorizeConfig      *DecolorizeConfig      `river:"decolorize,block,optional"`
	DockerConfig          *DockerConfig          `river:"docker,block,optional"`
	DropConfig            *DropConfig            `river:"drop,block,optional"`
	EventLogMessageConfig *EventLogMessageConfig `river:"eventlogmessage,block,optional"`
	GeoIPConfig           *GeoIPConfig           `river:"geoip,block,optional"`
	JSONConfig            *JSONConfig            `river:"json,block,optional"`
	LabelAllowConfig      *LabelAllowConfig      `river:"label_keep,block,optional"`
	LabelDropConfig       *LabelDropConfig       `river:"label_drop,block,optional"`
	LabelsConfig          *LabelsConfig          `river:"labels,block,optional"`
	LimitConfig           *LimitConfig           `river:"limit,block,optional"`
	LogfmtConfig          *LogfmtConfig          `river:"logfmt,block,optional"`
	MatchConfig           *MatchConfig           `river:"match,block,optional"`
	MetricsConfig         *MetricsConfig         `river:"metrics,block,optional"`
	MultilineConfig       *MultilineConfig       `river:"multiline,block,optional"`
	OutputConfig          *OutputConfig          `river:"output,block,optional"`
	PackConfig            *PackConfig            `river:"pack,block,optional"`
	RegexConfig           *RegexConfig           `river:"regex,block,optional"`
	ReplaceConfig         *ReplaceConfig         `river:"replace,block,optional"`
	StaticLabelsConfig    *StaticLabelsConfig    `river:"static_labels,block,optional"`
	StructuredMetadata    *LabelsConfig          `river:"structured_metadata,block,optional"`
	SamplingConfig        *SamplingConfig        `river:"sampling,block,optional"`
	TemplateConfig        *TemplateConfig        `river:"template,block,optional"`
	TenantConfig          *TenantConfig          `river:"tenant,block,optional"`
	TimestampConfig       *TimestampConfig       `river:"timestamp,block,optional"`
}

StageConfig defines a single stage in a processing pipeline. We define these as pointers types so we can use reflection to check that exactly one is set.

type StaticLabelsConfig 

type StaticLabelsConfig struct {
	Values map[string]*string `river:"values,attr"`
}

StaticLabelsConfig contains a map of static labels to be set.

type TemplateConfig 

type TemplateConfig struct {
	Source   string `river:"source,attr"`
	Template string `river:"template,attr"`
}

TemplateConfig configures template value extraction.

type TenantConfig 

type TenantConfig struct {
	Label  string `river:"label,attr,optional"`
	Source string `river:"source,attr,optional"`
	Value  string `river:"value,attr,optional"`
}

TenantConfig configures a tenant stage.

type TimestampConfig 

type TimestampConfig struct {
	Source          string   `river:"source,attr"`
	Format          string   `river:"format,attr"`
	FallbackFormats []string `river:"fallback_formats,attr,optional"`
	Location        *string  `river:"location,attr,optional"`
	ActionOnFailure string   `river:"action_on_failure,attr,optional"`
}

TimestampConfig configures a processing stage for timestamp extraction.

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.