row

Documentation

Index

• Variables
• type ActiveStats
  • func (as *ActiveStats) Done(n int, err error)
  • func (as *ActiveStats) GetStats() Stats
  • func (as *ActiveStats) Inc()
  • func (as *ActiveStats) MoveToPending(n int)
• type Annotatable
• type Base
  • func NewBase(label string, sink Sink, bufSize int, ann v2as.Annotator) *Base
  • func (pb *Base) Flush() error
  • func (pb *Base) GetStats() Stats
  • func (pb *Base) Put(row Annotatable) error
  • func (pb *Base) TaskError() error
• type Buffer
  • func NewBuffer(size int) *Buffer
  • func (buf *Buffer) Append(row interface{}) []interface{}
  • func (buf *Buffer) Reset() []interface{}
• type HasStats
• type NullAnnotator
  • func (row *NullAnnotator) AnnotateClients(map[string]*api.Annotations) error
  • func (row *NullAnnotator) AnnotateServer(*api.Annotations) error
  • func (row *NullAnnotator) GetClientIPs() []string
  • func (row *NullAnnotator) GetLogTime() time.Time
  • func (row *NullAnnotator) GetServerIP() string
• type Sink
• type Stats
  • func (s Stats) Total() int

Variables

var (
	ErrAnnotationError = errors.New("Annotation error")
	ErrNotAnnotatable  = errors.New("object does not implement Annotatable")
	ErrBufferFull      = errors.New("Buffer full")
	ErrInvalidSink     = errors.New("Not a valid row.Sink")
)

Errors that may be returned by Buffer functions.

Types

type ActiveStats

type ActiveStats struct {
	Stats
	// contains filtered or unexported fields
}

ActiveStats is a stats object that supports updates.

func (*ActiveStats) Done

func (as *ActiveStats) Done(n int, err error)

Done updates the pending to failed or committed.

func (*ActiveStats) GetStats

func (as *ActiveStats) GetStats() Stats

GetStats implements HasStats()

func (*ActiveStats) Inc

func (as *ActiveStats) Inc()

Inc increments the Buffered field

func (*ActiveStats) MoveToPending

func (as *ActiveStats) MoveToPending(n int)

MoveToPending increments the Pending field.

type Annotatable

type Annotatable interface {
	GetLogTime() time.Time
	GetClientIPs() []string // This is a slice to support mutliple hops in traceroute data.
	GetServerIP() string
	AnnotateClients(map[string]*api.Annotations) error // Must properly handle missing annotations.
	AnnotateServer(*api.Annotations) error             // Must properly handle nil parameter.
}

Annotatable interface enables integration of annotation into parser.Base. The row type should implement the interface, and the annotations will be added prior to insertion.

type Base

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

Base provides common parser functionality. Base is NOT THREAD-SAFE

func NewBase

func NewBase(label string, sink Sink, bufSize int, ann v2as.Annotator) *Base

NewBase creates a new Base. This will generally be embedded in a type specific parser.

func (*Base) Flush

func (pb *Base) Flush() error

Flush synchronously flushes any pending rows.

func (*Base) GetStats

func (pb *Base) GetStats() Stats

GetStats returns the buffer/sink stats.

func (*Base) Put

func (pb *Base) Put(row Annotatable) error

Put adds a row to the buffer. Iff the buffer is already full the prior buffered rows are annotated and committed to the Sink. NOTE: There is no guarantee about ordering of writes resulting from sequential calls to Put. However, once a block of rows is submitted to pb.commit, it should be written in the same order to the Sink. TODO improve Annotatable architecture.

func (*Base) TaskError

func (pb *Base) TaskError() error

TaskError return the task level error, based on failed rows, or any other criteria.

type Buffer

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

Buffer provides all basic functionality generally needed for buffering, annotating, and inserting rows that implement Annotatable. Buffer functions are THREAD-SAFE

func NewBuffer

func NewBuffer(size int) *Buffer

NewBuffer returns a new buffer of the desired size.

func (*Buffer) Append

func (buf *Buffer) Append(row interface{}) []interface{}

Append appends a row to the buffer. If buffer is full, this returns the buffered rows, and saves provided row in new buffer. Client MUST handle the returned rows.

func (*Buffer) Reset

func (buf *Buffer) Reset() []interface{}

Reset clears the buffer, returning all pending rows.

type HasStats

type HasStats interface {
	GetStats() Stats
}

HasStats can provide stats

type NullAnnotator

type NullAnnotator struct{}

NullAnnotator satisfies the Annotatable interface without actually doing anything.

func (*NullAnnotator) AnnotateClients

func (row *NullAnnotator) AnnotateClients(map[string]*api.Annotations) error

AnnotateClients does nothing.

func (*NullAnnotator) AnnotateServer

func (row *NullAnnotator) AnnotateServer(*api.Annotations) error

AnnotateServer does nothing.

func (*NullAnnotator) GetClientIPs

func (row *NullAnnotator) GetClientIPs() []string

GetClientIPs returns an empty array so nothing is annotated.

func (*NullAnnotator) GetLogTime

func (row *NullAnnotator) GetLogTime() time.Time

GetLogTime returns current time rather than the actual row time.

func (*NullAnnotator) GetServerIP

func (row *NullAnnotator) GetServerIP() string

GetServerIP returns an empty string because there is nothing to annotate.

type Sink

type Sink interface {
	Commit(rows []interface{}, label string) (int, error)
	io.Closer
}

Sink defines the interface for committing rows. Returns the number of rows successfully committed, and error. Implementations should be threadsafe.

type Stats

type Stats struct {
	Buffered  int // rows buffered but not yet sent.
	Pending   int // pending counts previously buffered rows that are being committed.
	Committed int
	Failed    int
}

Stats contains stats about buffer history.

func (Stats) Total

func (s Stats) Total() int

Total returns the total number of rows handled.