consumer

Documentation

Index

• Variables
• func MapperToSchema(mapper *coster.Mapper) bigquery.Schema
• type Aggregator
• type BigQueryAggregator
  • func NewBigQueryAggregator(ctx context.Context, project string, dataset string, table string, ...) (*BigQueryAggregator, error)
  • func (ba *BigQueryAggregator) Aggregate(ctx context.Context, ce coster.CostData) error
• type Consumer
• type CostRow
  • func (ce CostRow) Save() (row map[string]bigquery.Value, insertID string, err error)
• type PubsubConsumer
  • func NewPubsubConsumer(ctx context.Context, prometheusExporter *prometheus.Exporter, ...) (*PubsubConsumer, error)
  • func (pc *PubsubConsumer) Consume(ctx context.Context) error

Variables

var (
	// MeasureConsume measures aggregation results into the pubsub queue.
	MeasureConsume = stats.Int64("kostanza_aggregator/measures/consume", "Consumption operations", stats.UnitDimensionless)
	// TagConsumeStatus indicates the success or failure of a consumption
	TagConsumeStatus, _ = tag.NewKey("status")
)

Functions

func MapperToSchema

func MapperToSchema(mapper *coster.Mapper) bigquery.Schema

MapperToSchema creates a BigQuery schema representation for the provided coster.Mapper configuration.

Types

type Aggregator

type Aggregator interface {
	Aggregate(ctx context.Context, ce coster.CostData) error
}

Aggregator coalesces and persists coster.CostData from kostanza.

type BigQueryAggregator

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

BigQueryAggregator coalesces and persists coster.CosData data to BigQuery.

func NewBigQueryAggregator

func NewBigQueryAggregator(ctx context.Context, project string, dataset string, table string, mapper *coster.Mapper) (*BigQueryAggregator, error)

NewBigQueryAggregator creates a new Aggregator that publishes consumed pubsub events to the named BigQuery dataset and table. It will attempt to provision the table using a schema inferred from the current version of the application if the table does not yet exist.

func (*BigQueryAggregator) Aggregate

func (ba *BigQueryAggregator) Aggregate(ctx context.Context, ce coster.CostData) error

Aggregate pushes coster.CostData to BigQuery.

type Consumer

type Consumer interface {
	Consume(ctx context.Context) error
}

Consumer consumes messages as long as the provided context is not canceled, or its deadline exceeded.

type CostRow

type CostRow struct {
	coster.CostData
}

CostRow augments CostData with BigQuery specific interfaces for import purposes via the bigquery.Uploader.

func (CostRow) Save

func (ce CostRow) Save() (row map[string]bigquery.Value, insertID string, err error)

Save prepares a CostRow for import into BigQuery.

type PubsubConsumer

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

PubsubConsumer consumers messages from pubsub and forwards them to the provided aggregator.

func NewPubsubConsumer

func NewPubsubConsumer(ctx context.Context, prometheusExporter *prometheus.Exporter, listenAddr string, project string, topic string, subscription string, aggregator Aggregator) (*PubsubConsumer, error)

NewPubsubConsumer consumes messages from pubsub and invokes the provider aggregator with the message contents.

func (*PubsubConsumer) Consume

func (pc *PubsubConsumer) Consume(ctx context.Context) error

Consume begins the message consumption loop. It also registers and serves the `/metrics` and `/healthz` endpoints for monitoring purposes.