stage.go

package stages

import (
	"time"

	"github.com/go-kit/kit/log"
	"github.com/pkg/errors"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/common/model"

	"github.com/credativ/vali/pkg/valitail/api"
)

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"
	StageTypePack       = "pack"
	StageTypeLabelAllow = "labelallow"
)

// Processor takes an existing set of labels, timestamp and log entry and returns either a possibly mutated
// timestamp and log entry
type Processor interface {
	Process(labels model.LabelSet, extracted map[string]interface{}, time *time.Time, entry *string)
	Name() string
}

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

// Stage can receive entries via an inbound channel and forward mutated entries to an outbound channel.
type Stage interface {
	Name() string
	Run(chan Entry) chan Entry
}

// stageProcessor Allow to transform a Processor (old synchronous pipeline stage) into an async Stage
type stageProcessor struct {
	Processor
}

func (s stageProcessor) Run(in chan Entry) chan Entry {
	return RunWith(in, func(e Entry) Entry {
		s.Process(e.Labels, e.Extracted, &e.Timestamp, &e.Line)
		return e
	})
}

func toStage(p Processor) Stage {
	return &stageProcessor{Processor: p}
}

// New creates a new stage for the given type and configuration.
func New(logger log.Logger, jobName *string, stageType string,
	cfg interface{}, registerer prometheus.Registerer) (Stage, error) {
	var s Stage
	var err error
	switch stageType {
	case StageTypeDocker:
		s, err = NewDocker(logger, registerer)
		if err != nil {
			return nil, err
		}
	case StageTypeCRI:
		s, err = NewCRI(logger, registerer)
		if err != nil {
			return nil, err
		}
	case StageTypeJSON:
		s, err = newJSONStage(logger, cfg)
		if err != nil {
			return nil, err
		}
	case StageTypeRegex:
		s, err = newRegexStage(logger, cfg)
		if err != nil {
			return nil, err
		}
	case StageTypeMetric:
		s, err = newMetricStage(logger, cfg, registerer)
		if err != nil {
			return nil, err
		}
	case StageTypeLabel:
		s, err = newLabelStage(logger, cfg)
		if err != nil {
			return nil, err
		}
	case StageTypeLabelDrop:
		s, err = newLabelDropStage(cfg)
		if err != nil {
			return nil, err
		}
	case StageTypeTimestamp:
		s, err = newTimestampStage(logger, cfg)
		if err != nil {
			return nil, err
		}
	case StageTypeOutput:
		s, err = newOutputStage(logger, cfg)
		if err != nil {
			return nil, err
		}
	case StageTypeMatch:
		s, err = newMatcherStage(logger, jobName, cfg, registerer)
		if err != nil {
			return nil, err
		}
	case StageTypeTemplate:
		s, err = newTemplateStage(logger, cfg)
		if err != nil {
			return nil, err
		}
	case StageTypeTenant:
		s, err = newTenantStage(logger, cfg)
		if err != nil {
			return nil, err
		}
	case StageTypeReplace:
		s, err = newReplaceStage(logger, cfg)
		if err != nil {
			return nil, err
		}
	case StageTypeDrop:
		s, err = newDropStage(logger, cfg, registerer)
		if err != nil {
			return nil, err
		}
	case StageTypeMultiline:
		s, err = newMultilineStage(logger, cfg)
		if err != nil {
			return nil, err
		}
	case StageTypePack:
		s, err = newPackStage(logger, cfg, registerer)
		if err != nil {
			return nil, err
		}
	case StageTypeLabelAllow:
		s, err = newLabelAllowStage(cfg)
		if err != nil {
			return nil, err
		}
	default:
		return nil, errors.Errorf("Unknown stage type: %s", stageType)
	}
	return s, nil
}