stages

Documentation

Index

• Constants
• Variables
• func RunWith(input chan Entry, process func(e Entry) Entry) chan Entry
• type DropConfig
• type Entry
• type JSONConfig
• type LabelDropConfig
• type LabelsConfig
• type MatcherConfig
• type MetricConfig
• type MetricsConfig
• type MultilineConfig
• type OutputConfig
• 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 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 (
	ErrExpressionsRequired  = "JMES expression is required"
	ErrCouldNotCompileJMES  = "could not compile JMES expression"
	ErrEmptyJSONStageConfig = "empty json stage configuration"
	ErrEmptyJSONStageSource = "empty source"
)

Config Errors

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

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"
	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"
	StageTypeMultiline = "multiline"
)

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

Config Errors

const (
	ErrTenantStageEmptySourceOrValue        = "source or value config are required"
	ErrTenantStageConflictingSourceAndValue = "source and value are mutually exclusive: you should set source or value but not both"
)

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 (
	// ErrEmptyLabelDropStageConfig error returned if config is empty
	ErrEmptyLabelDropStageConfig = "labeldrop stage config cannot be empty"
)

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

const RFC3339Nano = "RFC3339Nano"

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
)

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.

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"`
}

JSONConfig represents a JSON Stage configuration

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 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 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 TemplateConfig

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

TemplateConfig configures template value extraction

type TenantConfig

type TenantConfig struct {
	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