fork

Documentation

Index

• type Distributor
• type GenericDistributor
  • func (d *GenericDistributor) Distribute(sample *bitflow.Sample, header *bitflow.Header) ([]Subpipeline, error)
  • func (d *GenericDistributor) String() string
• type Merger
  • func (sink *Merger) Close()
  • func (sink *Merger) Sample(sample *bitflow.Sample, header *bitflow.Header) error
  • func (sink *Merger) Start(wg *sync.WaitGroup) (_ golib.StopChan)
  • func (sink *Merger) String() string
• type MultiFileDistributor
  • func (b *MultiFileDistributor) Distribute(sample *bitflow.Sample, _ *bitflow.Header) ([]Subpipeline, error)
  • func (b *MultiFileDistributor) String() string
• type MultiMetricSource
  • func (in *MultiMetricSource) Add(subPipeline *bitflow.SamplePipeline)
  • func (in *MultiMetricSource) AddSource(source bitflow.SampleSource, steps ...bitflow.SampleProcessor)
  • func (in *MultiMetricSource) Close()
  • func (in *MultiMetricSource) ContainedStringers() []fmt.Stringer
  • func (in *MultiMetricSource) Start(wg *sync.WaitGroup) golib.StopChan
  • func (in *MultiMetricSource) String() string
• type MultiPipeline
  • func (m *MultiPipeline) Init(outgoing bitflow.SampleProcessor, closeHook func(), wg *sync.WaitGroup)
  • func (m *MultiPipeline) LogFinishedPipeline(isPassive bool, err error, prefix string)
  • func (m *MultiPipeline) StartPipeline(pipeline *bitflow.SamplePipeline, finishedHook func(isPassive bool, err error))
  • func (m *MultiPipeline) StopPipelines()
• type MultiplexDistributor
  • func (d *MultiplexDistributor) ContainedStringers() []fmt.Stringer
  • func (d *MultiplexDistributor) Distribute(_ *bitflow.Sample, _ *bitflow.Header) ([]Subpipeline, error)
  • func (d *MultiplexDistributor) String() string
• type PipelineArray
• type PipelineBuildFunc
• type PipelineCache
  • func (d *PipelineCache) ContainedStringers() []fmt.Stringer
• type RegexDistributor
  • func (d *RegexDistributor) ContainedStringers() []fmt.Stringer
  • func (d *RegexDistributor) Init() error
• type RoundRobinDistributor
  • func (rr *RoundRobinDistributor) ContainedStringers() []fmt.Stringer
  • func (rr *RoundRobinDistributor) Distribute(sample *bitflow.Sample, header *bitflow.Header) ([]Subpipeline, error)
  • func (rr *RoundRobinDistributor) String() string
  • func (rr *RoundRobinDistributor) TotalWeight() (res int)
• type SampleFork
  • func (f *SampleFork) Close()
  • func (f *SampleFork) ContainedStringers() []fmt.Stringer
  • func (f *SampleFork) Sample(sample *bitflow.Sample, header *bitflow.Header) error
  • func (f *SampleFork) Start(wg *sync.WaitGroup) golib.StopChan
  • func (f *SampleFork) String() string
• type Subpipeline
• type TagDistributor
  • func (d *TagDistributor) Distribute(sample *bitflow.Sample, _ *bitflow.Header) ([]Subpipeline, error)
  • func (d *TagDistributor) String() string

Types

type Distributor

type Distributor interface {
	Distribute(sample *bitflow.Sample, header *bitflow.Header) ([]Subpipeline, error)
	String() string
}

type GenericDistributor

type GenericDistributor struct {
	RegexDistributor
	GetKeys     func(sample *bitflow.Sample, header *bitflow.Header) []string
	Description string
}

func (*GenericDistributor) Distribute

func (d *GenericDistributor) Distribute(sample *bitflow.Sample, header *bitflow.Header) ([]Subpipeline, error)

func (*GenericDistributor) String

func (d *GenericDistributor) String() string

type Merger

type Merger struct {
	bitflow.AbstractSampleProcessor
	// contains filtered or unexported fields
}

func (*Merger) Close

func (sink *Merger) Close()

func (*Merger) Sample

func (sink *Merger) Sample(sample *bitflow.Sample, header *bitflow.Header) error

func (*Merger) Start

func (sink *Merger) Start(wg *sync.WaitGroup) (_ golib.StopChan)

func (*Merger) String

func (sink *Merger) String() string

type MultiFileDistributor

type MultiFileDistributor struct {
	bitflow.TagTemplate
	PipelineCache
	Config             bitflow.FileSink // Configuration parameters in this field will be used for file outputs
	ExtendSubpipelines func(fileName string, pipe *bitflow.SamplePipeline)
}

func (*MultiFileDistributor) Distribute

func (b *MultiFileDistributor) Distribute(sample *bitflow.Sample, _ *bitflow.Header) ([]Subpipeline, error)

func (*MultiFileDistributor) String

func (b *MultiFileDistributor) String() string

type MultiMetricSource

type MultiMetricSource struct {
	MultiPipeline
	bitflow.AbstractSampleProcessor
	// contains filtered or unexported fields
}

func (*MultiMetricSource) Add

func (in *MultiMetricSource) Add(subPipeline *bitflow.SamplePipeline)

func (*MultiMetricSource) AddSource

func (in *MultiMetricSource) AddSource(source bitflow.SampleSource, steps ...bitflow.SampleProcessor)

func (*MultiMetricSource) Close

func (in *MultiMetricSource) Close()

func (*MultiMetricSource) ContainedStringers

func (in *MultiMetricSource) ContainedStringers() []fmt.Stringer

func (*MultiMetricSource) Start

func (in *MultiMetricSource) Start(wg *sync.WaitGroup) golib.StopChan

func (*MultiMetricSource) String

func (in *MultiMetricSource) String() string

type MultiPipeline

type MultiPipeline struct {
	SequentialClose bool
	// contains filtered or unexported fields
}

func (*MultiPipeline) Init

func (m *MultiPipeline) Init(outgoing bitflow.SampleProcessor, closeHook func(), wg *sync.WaitGroup)

func (*MultiPipeline) LogFinishedPipeline

func (m *MultiPipeline) LogFinishedPipeline(isPassive bool, err error, prefix string)

func (*MultiPipeline) StartPipeline

func (m *MultiPipeline) StartPipeline(pipeline *bitflow.SamplePipeline, finishedHook func(isPassive bool, err error))

func (*MultiPipeline) StopPipelines

func (m *MultiPipeline) StopPipelines()

type MultiplexDistributor

type MultiplexDistributor struct {
	PipelineArray
}

func (*MultiplexDistributor) ContainedStringers

func (d *MultiplexDistributor) ContainedStringers() []fmt.Stringer

func (*MultiplexDistributor) Distribute

func (d *MultiplexDistributor) Distribute(_ *bitflow.Sample, _ *bitflow.Header) ([]Subpipeline, error)

func (*MultiplexDistributor) String

func (d *MultiplexDistributor) String() string

type PipelineArray

type PipelineArray struct {
	Subpipelines []*bitflow.SamplePipeline
	// contains filtered or unexported fields
}

type PipelineBuildFunc

type PipelineBuildFunc func(key string) ([]*bitflow.SamplePipeline, error)

type PipelineCache

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

func (*PipelineCache) ContainedStringers

func (d *PipelineCache) ContainedStringers() []fmt.Stringer

type RegexDistributor

type RegexDistributor struct {
	Pipelines map[string]func() ([]*bitflow.SamplePipeline, error)

	ExactMatch bool // Key patterns must match exactly, no glob (*) processing
	RegexMatch bool // Overrides ExactMatch -> treat key patterns as regexes
	// contains filtered or unexported fields
}

func (*RegexDistributor) ContainedStringers

func (d *RegexDistributor) ContainedStringers() []fmt.Stringer

func (*RegexDistributor) Init

func (d *RegexDistributor) Init() error

type RoundRobinDistributor

type RoundRobinDistributor struct {
	PipelineArray
	Weights []int // Optionally define weights for the pipelines (same order as pipelines). Only values >= 1 will be counted. Default weight is 1.
	// contains filtered or unexported fields
}

func (*RoundRobinDistributor) ContainedStringers

func (rr *RoundRobinDistributor) ContainedStringers() []fmt.Stringer

func (*RoundRobinDistributor) Distribute

func (rr *RoundRobinDistributor) Distribute(sample *bitflow.Sample, header *bitflow.Header) ([]Subpipeline, error)

func (*RoundRobinDistributor) String

func (rr *RoundRobinDistributor) String() string

func (*RoundRobinDistributor) TotalWeight

func (rr *RoundRobinDistributor) TotalWeight() (res int)

type SampleFork

type SampleFork struct {
	MultiPipeline
	bitflow.NoopProcessor

	Distributor Distributor

	// TODO implement fatal/non-fatal sub-pipeline errors
	// If true, errors of subpipelines will be logged but don't stop the entire MultiPipeline
	// Finished pipelines must be reported through LogFinishedPipeline()
	NonfatalErrors bool

	ForkPath []string
	// contains filtered or unexported fields
}

func (*SampleFork) Close

func (f *SampleFork) Close()

func (*SampleFork) ContainedStringers

func (f *SampleFork) ContainedStringers() []fmt.Stringer

func (*SampleFork) Sample

func (f *SampleFork) Sample(sample *bitflow.Sample, header *bitflow.Header) error

func (*SampleFork) Start

func (f *SampleFork) Start(wg *sync.WaitGroup) golib.StopChan

func (*SampleFork) String

func (f *SampleFork) String() string

type Subpipeline

type Subpipeline struct {
	Pipe *bitflow.SamplePipeline
	Key  string
}

type TagDistributor

type TagDistributor struct {
	RegexDistributor
	bitflow.TagTemplate
}

func (*TagDistributor) Distribute

func (d *TagDistributor) Distribute(sample *bitflow.Sample, _ *bitflow.Header) ([]Subpipeline, error)

func (*TagDistributor) String

func (d *TagDistributor) String() string