collector

Documentation

Index

• Constants
• Variables
• func AddChildLogger(logger *zap.SugaredLogger, d *processor.Document)
• func Collect(ctx context.Context, emitter Emitter, handleErr ErrHandler) error
• func DeregisterDocumentCollector(collectorType string)
• func Publish(ctx context.Context, d *processor.Document, blobStore *blob.BlobStore, ...) error
• func RegisterDocumentCollector(c Collector, collectorType string) error
• type Collector
• type DeregisterCollector
• type Emitter
• type ErrHandler

Constants

const (
	BufferChannelSize int = 1000
)

Variables

var (
	ErrCollectorOverwrite = fmt.Errorf("the document collector is being overwritten")
)

Functions

func AddChildLogger

func AddChildLogger(logger *zap.SugaredLogger, d *processor.Document)

func Collect

func Collect(ctx context.Context, emitter Emitter, handleErr ErrHandler) error

Collect takes all the collectors and starts collecting artifacts after Collect is called, no calls to RegisterDocumentCollector should happen.

func DeregisterDocumentCollector

func DeregisterDocumentCollector(collectorType string)

func Publish

func Publish(ctx context.Context, d *processor.Document, blobStore *blob.BlobStore, pubsub *emitter.EmitterPubSub, pubToQueue bool) error

Publish takes the "document" collected by the collectors and stores it into a blob store for retrieval by the processor/ingestor. A CDEvent is created to transmit the key (which is the sha256 of the collected "document"). This also fixes the issues where the "document" was too large to be sent across the event stream.

func RegisterDocumentCollector

func RegisterDocumentCollector(c Collector, collectorType string) error

Types

type Collector

type Collector interface {
	// RetrieveArtifacts collects the documents from the collector. It emits each collected
	// document through the channel to be collected and processed by the upstream processor.
	// The function should block until all the artifacts are collected and return a nil error
	// or return an error from the collector crashing. This function can keep running and check
	// for new artifacts as they are being uploaded by polling on an interval or run once and
	// grab all the artifacts and end.
	RetrieveArtifacts(ctx context.Context, docChannel chan<- *processor.Document) error
	// Type returns the collector type
	Type() string
}

type DeregisterCollector

type DeregisterCollector interface {
	DeregisterCollector(collectorType string) error
}

type Emitter

type Emitter func(*processor.Document) error

Emitter processes a document

type ErrHandler

type ErrHandler func(error) bool

ErrHandler processes an error and returns a boolean representing if the error was able to be gracefully handled