runner

Documentation

Overview

Package runner executes the mapper pipeline.

Index

• func DoMapLogItemsFn(fn MapLogItemFn, msgs []*mutator.LogMessage, emitErr func(error), ...) []*entry.IndexedValue
• func DoMapMapLeafFn(fn MapMapLeafFn, leaves []*tpb.MapLeaf, incFn IncMetricFn) ([]*entry.IndexedValue, error)
• func DoMapMetaFn(fn MapMetaFn, meta *spb.MapMetadata, incFn IncMetricFn) []*spb.MapMetadata_SourceSlice
• func DoMarshalIndexedValues(ivs <-chan *entry.IndexedValue, emitErr func(error), incFn IncMetricFn) []*tpb.MapLeaf
• func DoReadFn(ctx context.Context, fn ReadSliceFn, slices []*spb.MapMetadata_SourceSlice, ...) ([]*mutator.LogMessage, error)
• func DoReduceFn(reduceFn ReduceMutationFn, joined <-chan *Joined, emitErr func(error), ...) <-chan *entry.IndexedValue
• func Join(leaves []*entry.IndexedValue, msgs []*entry.IndexedValue, incFn IncMetricFn) <-chan *Joined
• type IncMetricFn
• type Joined
• type MapLogItemFn
• type MapMapLeafFn
• type MapMetaFn
• type ReadSliceFn
• type ReduceMutationFn

Functions

func DoMapLogItemsFn

func DoMapLogItemsFn(fn MapLogItemFn, msgs []*mutator.LogMessage,
	emitErr func(error), incFn IncMetricFn) []*entry.IndexedValue

DoMapLogItemsFn runs the MapLogItemsFn on each element of msgs.

func DoMapMapLeafFn

func DoMapMapLeafFn(fn MapMapLeafFn, leaves []*tpb.MapLeaf, incFn IncMetricFn) ([]*entry.IndexedValue, error)

DoMapMapLeafFn runs MapMapLeafFn on each MapLeaf.

func DoMapMetaFn

func DoMapMetaFn(fn MapMetaFn, meta *spb.MapMetadata, incFn IncMetricFn) []*spb.MapMetadata_SourceSlice

DoMapMetaFn runs MapMetaFn on meta and collects the outputs.

func DoMarshalIndexedValues

func DoMarshalIndexedValues(ivs <-chan *entry.IndexedValue, emitErr func(error), incFn IncMetricFn) []*tpb.MapLeaf

DoMarshalIndexedValues executes Marshal on each IndexedValue If marshal fails, it will emit an error and continue with a subset of ivs.

func DoReadFn

func DoReadFn(ctx context.Context, fn ReadSliceFn, slices []*spb.MapMetadata_SourceSlice,
	directoryID string, chunkSize int32, incFn IncMetricFn) ([]*mutator.LogMessage, error)

DoReadFn runs ReadSliceFn on every source slice and collects the outputs.

func DoReduceFn

func DoReduceFn(reduceFn ReduceMutationFn, joined <-chan *Joined, emitErr func(error),
	incFn IncMetricFn) <-chan *entry.IndexedValue

DoReduceFn takes the set of mutations and applies them to given leaves. Returns a channel of key value pairs that should be written to the map.

func Join

func Join(leaves []*entry.IndexedValue, msgs []*entry.IndexedValue, incFn IncMetricFn) <-chan *Joined

Join pairs up MapLeaves and IndexedValue by index.

Types

type IncMetricFn

type IncMetricFn func(label string)

IncMetricFn increments a metric

type Joined

type Joined struct {
	Index   []byte
	Values1 []*pb.EntryUpdate
	Values2 []*pb.EntryUpdate
}

Joined is the result of a CoGroupByKey on []*MapLeaf and []*IndexedValue.

type MapLogItemFn

type MapLogItemFn func(logItem *mutator.LogMessage,
	emit func(index []byte, mutation *pb.EntryUpdate), emitErr func(error))

MapLogItemFn takes a log item and emits 0 or more KV<index, mutations> pairs.

type MapMapLeafFn

type MapMapLeafFn func(*tpb.MapLeaf) (*entry.IndexedValue, error)

MapMapLeafFn converts an update into an IndexedValue.

type MapMetaFn

type MapMetaFn func(meta *spb.MapMetadata, emit func(*spb.MapMetadata_SourceSlice))

MapMetaFn emits a source slice for every map slice.

type ReadSliceFn

type ReadSliceFn func(ctx context.Context, slice *spb.MapMetadata_SourceSlice,
	directoryID string, chunkSize int32,
	emit func(*mutator.LogMessage)) error

ReadSliceFn emits the log messages referenced by slice.

type ReduceMutationFn

type ReduceMutationFn func(msgs []*pb.EntryUpdate, leaves []*pb.EntryUpdate,
	emit func(*pb.EntryUpdate), emitErr func(error))

ReduceMutationFn takes all the mutations for an index and an auxiliary input of existing mapleaf(s) and emits a new value for the index. ReduceMutationFn must be idempotent, commutative, and associative. i.e. must produce the same output regardless of input order or grouping, and it must be safe to run multiple times.