config

Documentation

Index

• Variables
• func ParseConfig() error
• type Encode
• type Extract
• type File
• type GenericMap
  • func (m GenericMap) Copy() GenericMap
• type Health
• type Ingest
• type Options
• type PipelineBuilderStage
  • func NewCollectorPipeline(name string, ingest api.IngestCollector) PipelineBuilderStage
  • func NewGRPCPipeline(name string, ingest api.IngestGRPCProto) PipelineBuilderStage
  • func NewKafkaPipeline(name string, ingest api.IngestKafka) PipelineBuilderStage
  • func (b *PipelineBuilderStage) Aggregate(name string, aggs []api.AggregateDefinition) PipelineBuilderStage
  • func (b *PipelineBuilderStage) EncodeKafka(name string, kafka api.EncodeKafka) PipelineBuilderStage
  • func (b *PipelineBuilderStage) EncodePrometheus(name string, prom api.PromEncode) PipelineBuilderStage
  • func (b *PipelineBuilderStage) GetStageParams() []StageParam
  • func (b *PipelineBuilderStage) GetStages() []Stage
  • func (b *PipelineBuilderStage) TransformFilter(name string, filter api.TransformFilter) PipelineBuilderStage
  • func (b *PipelineBuilderStage) TransformGeneric(name string, gen api.TransformGeneric) PipelineBuilderStage
  • func (b *PipelineBuilderStage) TransformNetwork(name string, nw api.TransformNetwork) PipelineBuilderStage
  • func (b *PipelineBuilderStage) WriteLoki(name string, loki api.WriteLoki) PipelineBuilderStage
  • func (b *PipelineBuilderStage) WriteStdout(name string, stdout api.WriteStdout) PipelineBuilderStage
• type Stage
• type StageParam
• type Transform
• type Write

Variables

var (
	Opt        = Options{}
	PipeLine   []Stage
	Parameters []StageParam
)

Functions

func ParseConfig

func ParseConfig() error

ParseConfig creates the internal unmarshalled representation from the Pipeline and Parameters json

Types

type Encode

type Encode struct {
	Type  string           `json:"type"`
	Prom  *api.PromEncode  `json:"prom,omitempty"`
	Kafka *api.EncodeKafka `json:"kafka,omitempty"`
}

type Extract

type Extract struct {
	Type       string                    `json:"type"`
	Aggregates []api.AggregateDefinition `json:"aggregates,omitempty"`
}

type File

type File struct {
	Filename string      `json:"filename"`
	Decoder  api.Decoder `json:"decoder"`
	Loop     bool        `json:"loop"`
	Chunks   int         `json:"chunks"`
}

type GenericMap

type GenericMap map[string]interface{}

func (GenericMap) Copy

func (m GenericMap) Copy() GenericMap

Copy will create a flat copy of GenericMap

type Health

type Health struct {
	Port string
}

type Ingest

type Ingest struct {
	Type      string               `json:"type"`
	File      *File                `json:"file,omitempty"`
	Collector *api.IngestCollector `json:"collector,omitempty"`
	Kafka     *api.IngestKafka     `json:"kafka,omitempty"`
	GRPC      *api.IngestGRPCProto `json:"grpc,omitempty"`
}

type Options

type Options struct {
	PipeLine   string
	Parameters string
	Health     Health
}

type PipelineBuilderStage

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

PipelineBuilderStage holds information about a created pipeline stage. This stage can be used to chain a following stage, or several of them (resulting in a fork). Example:

firstStage := NewCollectorPipeline("first stage", ...)
secondStage := firstStage.DecodeJSON("second stage")
thirdStage := secondStage.WriteLoki("third stage", ...)
forkedStage := secondStage.WriteStdout("fork following second stage", ...)

All created stages hold a pointer to the whole pipeline, so that the resulting pipeline can be retrieve from any of the stages:

forkedStage.GetStages()
forkedStage.GetStageParams()
// is equivalent to:
firstStage.GetStages()
firstStage.GetStageParams()

func NewCollectorPipeline

func NewCollectorPipeline(name string, ingest api.IngestCollector) PipelineBuilderStage

NewCollectorPipeline creates a new pipeline from an `IngestCollector` initial stage (listening for NetFlows / IPFIX)

func NewGRPCPipeline

func NewGRPCPipeline(name string, ingest api.IngestGRPCProto) PipelineBuilderStage

NewGRPCPipeline creates a new pipeline from an `IngestGRPCProto` initial stage (listening for NetObserv's eBPF agent protobuf)

func NewKafkaPipeline

func NewKafkaPipeline(name string, ingest api.IngestKafka) PipelineBuilderStage

NewKafkaPipeline creates a new pipeline from an `IngestKafka` initial stage (listening for flow events on Kafka)

func (*PipelineBuilderStage) Aggregate

func (b *PipelineBuilderStage) Aggregate(name string, aggs []api.AggregateDefinition) PipelineBuilderStage

Aggregate chains the current stage with an aggregate stage and returns that new stage

func (*PipelineBuilderStage) EncodeKafka

func (b *PipelineBuilderStage) EncodeKafka(name string, kafka api.EncodeKafka) PipelineBuilderStage

EncodeKafka chains the current stage with an EncodeKafka stage (writing to a Kafka topic) and returns that new stage

func (*PipelineBuilderStage) EncodePrometheus

func (b *PipelineBuilderStage) EncodePrometheus(name string, prom api.PromEncode) PipelineBuilderStage

EncodePrometheus chains the current stage with a PromEncode stage (to expose metrics in Prometheus format) and returns that new stage

func (*PipelineBuilderStage) GetStageParams

func (b *PipelineBuilderStage) GetStageParams() []StageParam

GetStageParams returns the current pipeline stage params. It can be called from any of the stages, they share the same pipeline reference.

func (*PipelineBuilderStage) GetStages

func (b *PipelineBuilderStage) GetStages() []Stage

GetStages returns the current pipeline stages. It can be called from any of the stages, they share the same pipeline reference.

func (*PipelineBuilderStage) TransformFilter

func (b *PipelineBuilderStage) TransformFilter(name string, filter api.TransformFilter) PipelineBuilderStage

TransformFilter chains the current stage with a TransformFilter stage and returns that new stage

func (*PipelineBuilderStage) TransformGeneric

func (b *PipelineBuilderStage) TransformGeneric(name string, gen api.TransformGeneric) PipelineBuilderStage

TransformGeneric chains the current stage with a TransformGeneric stage and returns that new stage

func (*PipelineBuilderStage) TransformNetwork

func (b *PipelineBuilderStage) TransformNetwork(name string, nw api.TransformNetwork) PipelineBuilderStage

TransformNetwork chains the current stage with a TransformNetwork stage and returns that new stage

func (*PipelineBuilderStage) WriteLoki

func (b *PipelineBuilderStage) WriteLoki(name string, loki api.WriteLoki) PipelineBuilderStage

WriteLoki chains the current stage with a WriteLoki stage and returns that new stage

func (*PipelineBuilderStage) WriteStdout

func (b *PipelineBuilderStage) WriteStdout(name string, stdout api.WriteStdout) PipelineBuilderStage

WriteStdout chains the current stage with a WriteStdout stage and returns that new stage

type Stage

type Stage struct {
	Name    string `json:"name"`
	Follows string `json:"follows,omitempty"`
}

type StageParam

type StageParam struct {
	Name      string     `json:"name"`
	Ingest    *Ingest    `json:"ingest,omitempty"`
	Transform *Transform `json:"transform,omitempty"`
	Extract   *Extract   `json:"extract,omitempty"`
	Encode    *Encode    `json:"encode,omitempty"`
	Write     *Write     `json:"write,omitempty"`
}

type Transform

type Transform struct {
	Type    string                `json:"type"`
	Generic *api.TransformGeneric `json:"generic,omitempty"`
	Filter  *api.TransformFilter  `json:"filter,omitempty"`
	Network *api.TransformNetwork `json:"network,omitempty"`
}

type Write

type Write struct {
	Type   string           `json:"type"`
	Loki   *api.WriteLoki   `json:"loki,omitempty"`
	Stdout *api.WriteStdout `json:"stdout,omitempty"`
}