bigquery

Documentation

Index

• func CreateDestination(ctx context.Context, config string) (eventkit.Destination, error)
• type Application
• type BatchQueue
  • func NewBatchQueue(target eventkit.Destination, queueSize int, batchSize int, ...) *BatchQueue
  • func (c *BatchQueue) Run(ctx context.Context)
  • func (c *BatchQueue) Submit(events ...*eventkit.Event)
• type BigQueryClient
  • func NewBigQueryClient(ctx context.Context, project string, datasetName string) (*BigQueryClient, error)
  • func (b *BigQueryClient) SaveRecord(ctx context.Context, records map[string][]*Record) error
• type BigQueryDestination
  • func NewBigQueryDestination(ctx context.Context, appName string, project string, dataset string) (*BigQueryDestination, error)
  • func (b *BigQueryDestination) Run(ctx context.Context)
  • func (b *BigQueryDestination) Submit(events ...*eventkit.Event)
• type BigQuerySink
  • func NewBigQuerySink(ctx context.Context, project string, dataset string) (*BigQuerySink, error)
  • func (b *BigQuerySink) Receive(ctx context.Context, unparsed *listener.Packet, packet pb.Packet) error
• type Parallel
  • func NewParallel(target func() (eventkit.Destination, error), workers int) *Parallel
  • func (p *Parallel) Run(ctx context.Context)
  • func (p *Parallel) Submit(events ...*eventkit.Event)
• type Record
  • func (r *Record) Save() (map[string]bigquery.Value, string, error)
• type Source

Functions

func CreateDestination

func CreateDestination(ctx context.Context, config string) (eventkit.Destination, error)

CreateDestination creates eventkit destination based on complex configuration. Example configurations:

127.0.0.1:1234
bigquery:app=...,project=...,dataset=...
bigquery:app=...,project=...,dataset=...|batch:queueSize=111,flashSize=111,flushInterval=111
bigquery:app=...,project=...,dataset=...|parallel:runners=10|batch:queueSize=111,flashSize=111,flushInterval=111

Types

type Application

type Application struct {
	Name    string
	Version string
}

type BatchQueue

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

BatchQueue collects events and send them in batches.

func NewBatchQueue

func NewBatchQueue(target eventkit.Destination, queueSize int, batchSize int, flushInterval time.Duration) *BatchQueue

NewBatchQueue creates a new batchQueue. It sends out the received events in batch. Either after the flushInterval is expired or when there are more than batchSize element in the queue.

func (*BatchQueue) Run

func (c *BatchQueue) Run(ctx context.Context)

Run implements Destination.

func (*BatchQueue) Submit

func (c *BatchQueue) Submit(events ...*eventkit.Event)

Submit implements Destination.

type BigQueryClient

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

func NewBigQueryClient

func NewBigQueryClient(ctx context.Context, project string, datasetName string) (*BigQueryClient, error)

func (*BigQueryClient) SaveRecord

func (b *BigQueryClient) SaveRecord(ctx context.Context, records map[string][]*Record) error

type BigQueryDestination

type BigQueryDestination struct {
	SourceInstance string
	// contains filtered or unexported fields
}

BigQueryDestination can be used to save each evenkit package directly to server.

func NewBigQueryDestination

func NewBigQueryDestination(ctx context.Context, appName string, project string, dataset string) (*BigQueryDestination, error)

func (*BigQueryDestination) Run

func (b *BigQueryDestination) Run(ctx context.Context)

func (*BigQueryDestination) Submit

func (b *BigQueryDestination) Submit(events ...*eventkit.Event)

Submit implements Destination.

type BigQuerySink

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

BigQuerySink provides an abstraction for processing events in a transport agnostic way.

func NewBigQuerySink

func NewBigQuerySink(ctx context.Context, project string, dataset string) (*BigQuerySink, error)

func (*BigQuerySink) Receive

func (b *BigQuerySink) Receive(ctx context.Context, unparsed *listener.Packet, packet pb.Packet) error

Receive is called when the server receive an event to process.

type Parallel

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

Parallel sends messages parallel from multiple goroutines.

func NewParallel

func NewParallel(target func() (eventkit.Destination, error), workers int) *Parallel

NewParallel creates a destination. It requires a way to create the worker destinations and the number of goroutines.

func (*Parallel) Run

func (p *Parallel) Run(ctx context.Context)

Run implements eventkit.Destination.

func (*Parallel) Submit

func (p *Parallel) Submit(events ...*eventkit.Event)

Submit implements eventkit.Destination.

type Record

type Record struct {
	Application Application

	Source Source

	ReceivedAt time.Time
	Timestamp  time.Time
	Correction time.Duration

	Tags []*pb.Tag
}

func (*Record) Save

func (r *Record) Save() (map[string]bigquery.Value, string, error)

type Source

type Source struct {
	Instance string
	Address  string
}