aggregator

README

Data Aggregator

Overview

The aggregator is a aggregation alias that allows developers to plug and play different data aggregation strategies based on the type of data they are working with. The aggregator maintains the latest data i.e. price for a given asset pair for each provider - whether its a validator, API provider, or other. When a aggregated information is requested, the price aggregator will utilize its configured strategies to determine the final result to return to the caller. In the case of price aggregation, the aggregator may return something like the median price across all providers.

NOTE: Each strategy must be deterministic if used in a distributed environment. This means that the same inputs must always produce the same outputs.

Configuration

The aggregator is configured by supplementing a data aggregator strategy.

// DataAggregator is a simple aggregator for provider data. It is thread-safe since
// it is assumed to be called concurrently in data fetching goroutines.
type DataAggregator[K comparable, V any] struct {
	mtx sync.RWMutex

	// aggregateFn is the function used to aggregate data from each provider.
	aggregateFn AggregateFn[K, V]

	// providerData is a map of provider -> value (i.e. prices).
	providerData AggregatedProviderData[K, V]

	// aggregatedData is the current set of aggregated data across the providers.
	aggregatedData V
}

The aggregation strategy is defined as follows:

// AggregateFn is the function used to aggregate data from each provider. Given a
// map of provider -> values, the aggregate function should return a final
// value.
AggregateFn[K comparable, V any] func(providers AggregatedProviderData[K, V]) V

// AggregateFnFromContext is a function that is used to parametrize an aggregateFn
// by an sdk.Context. This is used to allow the aggregateFn to access the latest state
// of an application i.e computing a stake weighted median based on the latest validator set.
AggregateFnFromContext[K comparable, V any] func(ctx sdk.Context) AggregateFn[K, V]

AggregateFn inputs data from each provider and outputs a final value. If developer's need to implement a strategy that requires stateful information - such as SDK state - they can utilize the AggregateFnFromContext type.

Please reference the sample implementation - ComputeMedian - for an example of how to implement a strategy.

Documentation

Index

• type AggregateFn
• type AggregateFnFromContext
• type AggregatedProviderData
• type Aggregator
• type DataAggregator
  • func NewDataAggregator[K comparable, V any](opts ...DataAggregatorOption[K, V]) *DataAggregator[K, V]
  • func (p *DataAggregator[K, V]) AggregateData()
  • func (p *DataAggregator[K, V]) AggregateDataFromContext(ctx sdk.Context)
  • func (p *DataAggregator[K, V]) GetAggregatedData() V
  • func (p *DataAggregator[K, V]) GetDataByProvider(provider K) V
  • func (p *DataAggregator[K, V]) GetProviderData() AggregatedProviderData[K, V]
  • func (p *DataAggregator[K, V]) ResetProviderData()
  • func (p *DataAggregator[K, V]) SetAggregatedData(aggregatedData V)
  • func (p *DataAggregator[K, V]) SetProviderData(provider K, data V)
• type DataAggregatorOption
  • func WithAggregateFn[K comparable, V any](fn AggregateFn[K, V]) DataAggregatorOption[K, V]
  • func WithAggregateFnFromContext[K comparable, V any](fn AggregateFnFromContext[K, V]) DataAggregatorOption[K, V]

Types

type AggregateFn

type AggregateFn[K comparable, V any] func(providers AggregatedProviderData[K, V]) V

AggregateFn is the function used to aggregate data from each provider. Given a map of provider -> values, the aggregate function should return a final value.

type AggregateFnFromContext

type AggregateFnFromContext[K comparable, V any] func(ctx sdk.Context) AggregateFn[K, V]

AggregateFnFromContext is a function that is used to parametrize an aggregateFn by an sdk.Context. This is used to allow the aggregateFn to access the latest state of an application i.e computing a stake weighted median based on the latest validator set.

type AggregatedProviderData

type AggregatedProviderData[K comparable, V any] map[K]V

AggregatedProviderData defines a type alias for a map of provider -> data (i.e. a set of prices).

type Aggregator

type Aggregator[K comparable, V any] interface {
	GetProviderData() AggregatedProviderData[K, V]
	GetDataByProvider(provider K) V
	SetProviderData(provider K, data V)
	ResetProviderData()
	AggregateData()
	AggregateDataFromContext(ctx sdk.Context)
	GetAggregatedData() V
	SetAggregatedData(aggregatedData V)
}

Aggregator defines the expected interface that must be implemented by any custom data aggregator.

type DataAggregator

type DataAggregator[K comparable, V any] struct {
	sync.Mutex
	// contains filtered or unexported fields
}

DataAggregator is a simple aggregator for provider data. It is thread-safe since it is assumed to be called concurrently in data fetching goroutines. The DataAggregator requires one of either an aggregateFn or aggregateFnFromContext to be set.

func NewDataAggregator

func NewDataAggregator[K comparable, V any](opts ...DataAggregatorOption[K, V]) *DataAggregator[K, V]

NewDataAggregator returns a DataAggregator. The DataAggregator is responsible for aggregating data (such as prices) from each provider and computing the final aggregated data (final price). The DataAggregator is thread-safe since it is assumed to be called concurrently in price fetching goroutines.

func (*DataAggregator[K, V]) AggregateData

func (p *DataAggregator[K, V]) AggregateData()

AggregateData aggregates the current set of data by using the aggregate function.

func (*DataAggregator[K, V]) AggregateDataFromContext

func (p *DataAggregator[K, V]) AggregateDataFromContext(ctx sdk.Context)

AggregateDataFromContext aggregates the current set of data by using the aggregate function parametrized by the given context.

func (*DataAggregator[K, V]) GetAggregatedData

func (p *DataAggregator[K, V]) GetAggregatedData() V

GetAggregatedData returns the aggregated data based on the provided data.

func (*DataAggregator[K, V]) GetDataByProvider

func (p *DataAggregator[K, V]) GetDataByProvider(provider K) V

GetDataByProvider returns the data currently stored for a given provider.

func (*DataAggregator[K, V]) GetProviderData

func (p *DataAggregator[K, V]) GetProviderData() AggregatedProviderData[K, V]

GetProviderData returns a copy of the aggregated provider data.

func (*DataAggregator[K, V]) ResetProviderData

func (p *DataAggregator[K, V]) ResetProviderData()

ResetProviderData resets the data aggregator for all providers.

func (*DataAggregator[K, V]) SetAggregatedData

func (p *DataAggregator[K, V]) SetAggregatedData(aggregatedData V)

SetAggregatedData sets the current set of aggregated data.

func (*DataAggregator[K, V]) SetProviderData

func (p *DataAggregator[K, V]) SetProviderData(provider K, data V)

SetProviderData updates the data aggregator with the given provider and data.

type DataAggregatorOption

type DataAggregatorOption[K comparable, V any] func(*DataAggregator[K, V])

DataAggregatorOption is a function that is used to parametrize a DataAggregator instance.

func WithAggregateFn

func WithAggregateFn[K comparable, V any](fn AggregateFn[K, V]) DataAggregatorOption[K, V]

WithAggregateFn sets the aggregateFn of a DataAggregatorOptions.

func WithAggregateFnFromContext

func WithAggregateFnFromContext[K comparable, V any](fn AggregateFnFromContext[K, V]) DataAggregatorOption[K, V]

WithAggregateFnFromContext sets the aggregateFnFromContext of a DataAggregatorOptions.