stages

Documentation

Index

• Constants
• Variables
• func RunWith(input chan Entry, process func(e Entry) Entry) chan Entry
• func RunWithSkip(input chan Entry, process func(e Entry) (Entry, bool)) chan Entry
• func SetReadLineRateLimiter(rateVal float64, burstVal int, drop bool)
• type DropConfig
• type Entry
• type JSONConfig
• type LabelAllowConfig
• type LabelDropConfig
• type LabelsConfig
• type LimitConfig
• type LogfmtConfig
• type MatcherConfig
• type MetricConfig
• type MetricsConfig
• type MultilineConfig
• type OutputConfig
• type PackConfig
• type Packed
  • func (w Packed) MarshalJSON() ([]byte, error)
  • func (w *Packed) UnmarshalJSON(data []byte) error
• type Pipeline
  • func NewPipeline(logger log.Logger, stgs PipelineStages, jobName *string, ...) (*Pipeline, error)
  • func (p *Pipeline) Name() string
  • func (p *Pipeline) Run(in chan Entry) chan Entry
  • func (p *Pipeline) Size() int
  • func (p *Pipeline) Wrap(next api.EntryHandler) api.EntryHandler
• type PipelineStage
• type PipelineStages
• type Processor
• type RegexConfig
• type ReplaceConfig
• type Stage
  • func New(logger log.Logger, jobName *string, stageType string, cfg interface{}, ...) (Stage, error)
  • func NewCRI(logger log.Logger, registerer prometheus.Registerer) (Stage, error)
  • func NewDocker(logger log.Logger, registerer prometheus.Registerer) (Stage, error)
• type StaticLabelConfig
• type StaticLabelStage
  • func (l *StaticLabelStage) Name() string
  • func (l *StaticLabelStage) Process(labels model.LabelSet, extracted map[string]interface{}, t *time.Time, ...)
• type TemplateConfig
• type TenantConfig
• type TimestampConfig

Constants

const (
	ErrDropStageEmptyConfig     = "drop stage config must contain at least one of `source`, `expression`, `older_than` or `longer_than`"
	ErrDropStageInvalidDuration = "drop stage invalid duration, %v cannot be converted to a duration: %v"
	ErrDropStageInvalidConfig   = "drop stage config error, `value` and `expression` cannot both be defined at the same time."
	ErrDropStageInvalidRegex    = "drop stage regex compilation error: %v"
	ErrDropStageInvalidByteSize = "drop stage failed to parse longer_than to bytes: %v"
)

const (
	RFC3339Nano         = "RFC3339Nano"
	MaxPartialLinesSize = 100 // Max buffer size to hold partial lines.
)

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

const (
	ErrEmptyLabelStageConfig = "label stage config cannot be empty"
	ErrInvalidLabelName      = "invalid label name: %s"
)

const (
	ErrMappingRequired        = "logfmt mapping is required"
	ErrEmptyLogfmtStageConfig = "empty logfmt stage configuration"
	ErrEmptyLogfmtStageSource = "empty source"
)

Config Errors

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

const (
	MetricTypeCounter   = "counter"
	MetricTypeGauge     = "gauge"
	MetricTypeHistogram = "histogram"

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

const (
	ErrMultilineStageEmptyConfig        = "multiline stage config must define `firstline` regular expression"
	ErrMultilineStageInvalidRegex       = "multiline stage first line regex compilation error: %v"
	ErrMultilineStageInvalidMaxWaitTime = "multiline stage `max_wait_time` parse error: %v"
)

const (
	ErrEmptyOutputStageConfig = "output stage config cannot be empty"
	ErrOutputSourceRequired   = "output source value is required if output is specified"
)

Config Errors

const (
	ErrExpressionRequired    = "expression is required"
	ErrCouldNotCompileRegex  = "could not compile regular expression"
	ErrEmptyRegexStageConfig = "empty regex stage configuration"
	ErrEmptyRegexStageSource = "empty source"
)

Config Errors

const (
	ErrEmptyReplaceStageConfig = "empty replace stage configuration"
	ErrEmptyReplaceStageSource = "empty source in replace stage"
)

Config Errors

const (
	StageTypeJSON         = "json"
	StageTypeLogfmt       = "logfmt"
	StageTypeRegex        = "regex"
	StageTypeReplace      = "replace"
	StageTypeMetric       = "metrics"
	StageTypeLabel        = "labels"
	StageTypeLabelDrop    = "labeldrop"
	StageTypeTimestamp    = "timestamp"
	StageTypeOutput       = "output"
	StageTypeDocker       = "docker"
	StageTypeCRI          = "cri"
	StageTypeMatch        = "match"
	StageTypeTemplate     = "template"
	StageTypePipeline     = "pipeline"
	StageTypeTenant       = "tenant"
	StageTypeDrop         = "drop"
	StageTypeLimit        = "limit"
	StageTypeMultiline    = "multiline"
	StageTypePack         = "pack"
	StageTypeLabelAllow   = "labelallow"
	StageTypeStaticLabels = "static_labels"
)

const (
	ErrEmptyTemplateStageConfig = "template stage config cannot be empty"
	ErrTemplateSourceRequired   = "template source value is required"
)

Config Errors

const (
	ErrTenantStageEmptyLabelSourceOrValue        = "label, source or value config are required"
	ErrTenantStageConflictingLabelSourceAndValue = "label, source and value are mutually exclusive: you should set source, value or label but not all"
)

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

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

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

const (
	// ErrEmptyLabelAllowStageConfig error returned if config is empty
	ErrEmptyLabelAllowStageConfig = "labelallow stage config cannot be empty"
)

const (
	// ErrEmptyLabelDropStageConfig error returned if config is empty
	ErrEmptyLabelDropStageConfig = "labeldrop stage config cannot be empty"
)

const (
	// ErrEmptyStaticLabelStageConfig error returned if config is empty
	ErrEmptyStaticLabelStageConfig = "static_labels stage config cannot be empty"
)

const (
	ErrLimitStageInvalidRateOrBurst = "limit stage failed to parse rate or burst"
)

const (
	ErrTimestampContainsYear = "timestamp '%s' is expected to not contain the year date component"
)

Variables

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
)

var (
	TimestampActionOnFailureOptions = []string{TimestampActionOnFailureSkip, TimestampActionOnFailureFudge}
)

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 SetReadLineRateLimiter

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

Types

type DropConfig

type DropConfig struct {
	DropReason *string `mapstructure:"drop_counter_reason"`
	Source     *string `mapstructure:"source"`
	Value      *string `mapstructure:"value"`
	Expression *string `mapstructure:"expression"`

	OlderThan *string `mapstructure:"older_than"`

	LongerThan *string `mapstructure:"longer_than"`
	// contains filtered or unexported fields
}

DropConfig contains the configuration for a dropStage

type Entry

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

type JSONConfig

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

JSONConfig represents a JSON Stage configuration

type LabelAllowConfig

type LabelAllowConfig []string

labelallowConfig is a slice of labels to be included

type LabelDropConfig

type LabelDropConfig []string

LabelDropConfig is a slice of labels to be dropped

type LabelsConfig

type LabelsConfig map[string]*string

LabelsConfig is a set of labels to be extracted

type LimitConfig

type LimitConfig struct {
	Rate  float64 `mapstructure:"rate"`
	Burst int     `mapstructure:"burst"`
	Drop  bool    `mapstructure:"drop"`
}

type LogfmtConfig

type LogfmtConfig struct {
	Mapping map[string]string `mapstructure:"mapping"`
	Source  *string           `mapstructure:"source"`
}

LogfmtConfig represents a logfmt Stage configuration

type MatcherConfig

type MatcherConfig struct {
	PipelineName *string        `mapstructure:"pipeline_name"`
	Selector     string         `mapstructure:"selector"`
	Stages       PipelineStages `mapstructure:"stages"`
	Action       string         `mapstructure:"action"`
	DropReason   *string        `mapstructure:"drop_counter_reason"`
}

MatcherConfig contains the configuration for a matcherStage

type MetricConfig

type MetricConfig struct {
	MetricType   string  `mapstructure:"type"`
	Description  string  `mapstructure:"description"`
	Source       *string `mapstructure:"source"`
	Prefix       string  `mapstructure:"prefix"`
	IdleDuration *string `mapstructure:"max_idle_duration"`

	Config interface{} `mapstructure:"config"`
	// contains filtered or unexported fields
}

MetricConfig is a single metrics configuration.

type MetricsConfig

type MetricsConfig map[string]MetricConfig

MetricsConfig is a set of configured metrics.

type MultilineConfig

type MultilineConfig struct {
	Expression *string `mapstructure:"firstline"`

	MaxLines    *uint64 `mapstructure:"max_lines"`
	MaxWaitTime *string `mapstructure:"max_wait_time"`
	// contains filtered or unexported fields
}

MultilineConfig contains the configuration for a multilineStage

type OutputConfig

type OutputConfig struct {
	Source string `mapstructure:"source"`
}

OutputConfig configures output value extraction

type PackConfig

type PackConfig struct {
	Labels          []string `mapstrcuture:"labels"`
	IngestTimestamp *bool    `mapstructure:"ingest_timestamp"`
}

PackConfig contains the configuration for a packStage

type Packed

type Packed struct {
	Labels map[string]string `json:",inline"`
	Entry  string            `json:"_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, stgs PipelineStages, 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 api.EntryHandler) api.EntryHandler

Wrap implements EntryMiddleware

type PipelineStage

type PipelineStage = map[interface{}]interface{}

PipelineStage contains configuration for a single pipeline stage

type PipelineStages

type PipelineStages = []interface{}

PipelineStages contains configuration for each stage within a pipeline

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  `mapstructure:"expression"`
	Source     *string `mapstructure:"source"`
}

RegexConfig contains a regexStage configuration

type ReplaceConfig

type ReplaceConfig struct {
	Expression string  `mapstructure:"expression"`
	Source     *string `mapstructure:"source"`
	Replace    string  `mapstructure:"replace"`
}

ReplaceConfig contains a regexStage configuration

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, stageType string,
	cfg interface{}, registerer prometheus.Registerer) (Stage, error)

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

func NewCRI

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

NewCRI creates a CRI format specific pipeline stage

func NewDocker

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

NewDocker creates a Docker json log format specific pipeline stage.

type StaticLabelConfig

type StaticLabelConfig map[string]*string

StaticLabelConfig is a slice of static-labels to be included

type StaticLabelStage

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

func (*StaticLabelStage) Name

func (l *StaticLabelStage) Name() string

Name implements Stage

func (*StaticLabelStage) Process

func (l *StaticLabelStage) Process(labels model.LabelSet, extracted map[string]interface{}, t *time.Time, entry *string)

Process implements Stage

type TemplateConfig

type TemplateConfig struct {
	Source   string `mapstructure:"source"`
	Template string `mapstructure:"template"`
}

TemplateConfig configures template value extraction

type TenantConfig

type TenantConfig struct {
	Label  string `mapstructure:"label"`
	Source string `mapstructure:"source"`
	Value  string `mapstructure:"value"`
}

type TimestampConfig

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

TimestampConfig configures timestamp extraction