stages

Documentation

Index

• Constants
• Variables
• type DropConfig
• type JSONConfig
• type LabelsConfig
• type MatcherConfig
• type MetricConfig
• type MetricsConfig
• type OutputConfig
• type Pipeline
  • func NewPipeline(logger log.Logger, stgs PipelineStages, jobName *string, ...) (*Pipeline, error)
  • func (p *Pipeline) AddStage(stage Stage)
  • func (p *Pipeline) Name() string
  • func (p *Pipeline) Process(labels model.LabelSet, extracted map[string]interface{}, ts *time.Time, ...)
  • func (p *Pipeline) Size() int
  • func (p *Pipeline) Wrap(next api.EntryHandler) api.EntryHandler
• type PipelineStage
• type PipelineStages
• 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 StageFunc
  • func (s StageFunc) Process(labels model.LabelSet, extracted map[string]interface{}, time *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 (
	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 (
	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"
	StageTypeTimestamp = "timestamp"
	StageTypeOutput    = "output"
	StageTypeDocker    = "docker"
	StageTypeCRI       = "cri"
	StageTypeMatch     = "match"
	StageTypeTemplate  = "template"
	StageTypePipeline  = "pipeline"
	StageTypeTenant    = "tenant"
	StageTypeDrop      = "drop"
)

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 (
	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}
)

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 JSONConfig

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

JSONConfig represents a JSON Stage configuration

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 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) AddStage

func (p *Pipeline) AddStage(stage Stage)

AddStage adds a stage to the pipeline

func (*Pipeline) Name

func (p *Pipeline) Name() string

Name implements Stage

func (*Pipeline) Process

func (p *Pipeline) Process(labels model.LabelSet, extracted map[string]interface{}, ts *time.Time, entry *string)

Process implements Stage allowing a pipeline stage to also be an entire pipeline

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 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 {
	Process(labels model.LabelSet, extracted map[string]interface{}, time *time.Time, entry *string)
	Name() string
}

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

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 StageFunc

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

StageFunc is modelled on http.HandlerFunc.

func (StageFunc) Process

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

Process implements EntryHandler.

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"`
	Location        *string `mapstructure:"location"`
	ActionOnFailure *string `mapstructure:"action_on_failure"`
}

TimestampConfig configures timestamp extraction