segments

Documentation

Overview

This package is home to all pipeline segment implementations. Generally, every segment lives in its own package, implements the Segment interface, embeds the BaseSegment to take care of the I/O side of things, and has an additional init() function to register itself using RegisterSegment.

Index

• Variables
• func RegisterSegment(name string, s Segment)
• func TestSegment(name string, config map[string]string, msg *pb.EnrichedFlow) *pb.EnrichedFlow
• type BaseFilterSegment
  • func (segment *BaseFilterSegment) SubscribeDrops(drops chan<- *pb.EnrichedFlow)
• type BaseSegment
  • func (segment *BaseSegment) Rewire(in chan *pb.EnrichedFlow, out chan *pb.EnrichedFlow)
  • func (segment *BaseSegment) ShutdownParentPipeline()
• type Segment
  • func LookupSegment(name string) Segment

Variables

var (
	ContainerVolumePrefix = ""
)

Functions

func RegisterSegment

func RegisterSegment(name string, s Segment)

Used by Segments to register themselves in their init() functions. Errors and exits immediately on conflicts.

func TestSegment

func TestSegment(name string, config map[string]string, msg *pb.EnrichedFlow) *pb.EnrichedFlow

Used by the tests to run single flow messages through a segment.

Types

type BaseFilterSegment

type BaseFilterSegment struct {
	BaseSegment
	Drops chan<- *pb.EnrichedFlow
}

An extended basis for Segment implementations in the filter group. It contains the necessities to process filtered (dropped) flows.

func (*BaseFilterSegment) SubscribeDrops

func (segment *BaseFilterSegment) SubscribeDrops(drops chan<- *pb.EnrichedFlow)

Set a return channel for dropped flow messages. Segments need to be wary of this channel closing when producing messages to this channel. This method is only called by the flowpipeline tool from the controlflow/branch segment to implement the then/else branches, otherwise this functionality is unused.

type BaseSegment

type BaseSegment struct {
	In  <-chan *pb.EnrichedFlow
	Out chan<- *pb.EnrichedFlow
}

Serves as a basis for any Segment implementations. Segments embedding this type only need the New and the Run methods to be compliant to the Segment interface.

func (*BaseSegment) Rewire

func (segment *BaseSegment) Rewire(in chan *pb.EnrichedFlow, out chan *pb.EnrichedFlow)

This function rewires this Segment with the provided channels. This is typically called only by pipeline.New() and present in any Segment implementation embedding the BaseSegment. The peculiar implementation of passing the full channel list and providing indexes is due to the fact that controlflow segments may want to skip segments and thus need to have all later references available as well.

func (*BaseSegment) ShutdownParentPipeline

func (segment *BaseSegment) ShutdownParentPipeline()

This functions shutdown Parent Pipeline segments on the given syscall. It is used for intended termination within pipeline function, e.g. end pipeline on read from file.

type Segment

type Segment interface {
	New(config map[string]string) Segment                       // for reading the provided config
	Run(wg *sync.WaitGroup)                                     // goroutine, must close(segment.Out) when segment.In is closed
	Rewire(in chan *pb.EnrichedFlow, out chan *pb.EnrichedFlow) // embed this using BaseSegment
	ShutdownParentPipeline()                                    // shut down Parent Pipeline gracefully
}

This interface is central to an Pipeline object, as it operates on a list of them. In general, Segments should embed the BaseSegment to provide the Rewire function and the associated vars.

func LookupSegment

func LookupSegment(name string) Segment

Used by the pipeline package to convert segment names in configuration to actual Segment objects.