metrics

Documentation

Overview

A token bucket implementation that supports token debt Token debt is negative tokens that must be repaid before any more tokens can be taken Token debt is repaid by the passage of time

Throughput package tracks and stores relay-per-second data for each application It shares throughput reports through pubsub and aggregates all reports from other middleware instances

Index

• Variables
• func ValidateThroughputLimitConfig(config ThroughputLimitConfig) error
• type ConnectDone
• type ConnectStart
• type ConnectionReport
• type DNSDone
• type DNSReport
• type DNSStart
• type GotConnection
• type Key
  • func (k Key) String() string
• type Limiter
  • func NewTokenBucket(rate int64, maxTokens int64) Limiter
• type NetworkReport
  • func (n *NetworkReport) String() string
• type NetworkTracer
  • func NewNetworkTracer(config NetworkTracerConfig, logger *logger.Logger) NetworkTracer
• type NetworkTracerConfig
  • func (n NetworkTracerConfig) IsEmpty() bool
• type NewLimiter
• type PortalAppGetter
• type Recorder
  • func NewMetricsRecorder(nc, gnc *nats.Conn, config RecorderConfig, samplesChan chan Sample, ...) (Recorder, error)
• type RecorderConfig
  • func (r *RecorderConfig) SetDefaults()
• type Relay
  • func (meta *Relay) LoadFromLog(r relay.RelayLog)
  • func (meta *Relay) String() string
• type RelayRequest
• type RelayResult
• type RoundTripReport
• type Sample
  • func GetSampleFromRelay(relay Relay) Sample
  • func (s *Sample) String() string
• type ServerReport
• type Stats
• type TLSReport
• type ThroughputAllower
• type ThroughputLimitConfig
• type ThroughputMetric
• type ThroughputReport
• type ThroughputStore
  • func NewThroughputStore(sendChan chan ThroughputReport, recvChan chan ThroughputReport, ...) *ThroughputStore
  • func (t *ThroughputStore) IsRelayAllowed(appId string, limit int32, count int) (allowed bool)
  • func (t *ThroughputStore) Start()
• type TokenBucket
  • func (t *TokenBucket) Allow(timestamp time.Time, count int64) bool
  • func (t *TokenBucket) Take(timestamp time.Time, tokens int64)

Variables

var (
	ErrApplicationNotFound              = errors.New("Application not found")
	ErrThroughputLimitConfigRate        = errors.New("Throughput rate must be specified and above 0")
	ErrThroughputLimitSendInterval      = errors.New("Throughput report send interval must be specified and above 10")
	ErrThroughputRateMetricMinimum      = errors.New("Throughput Rate Metric Minimum must be above 0")
	ErrThroughputRateTrackerHistorySize = errors.New("Throughput Rate Tracker History Size must be greater or equal to 5")
)

var (
	ErrTimestampNotFound = fmt.Errorf("Timestamp not found")
)

Functions

func ValidateThroughputLimitConfig

func ValidateThroughputLimitConfig(config ThroughputLimitConfig) error

Types

type ConnectDone

type ConnectDone struct {
	Timestamp time.Time
	HadError  bool
	Address   string
}

type ConnectStart

type ConnectStart struct {
	Timestamp time.Time
}

type ConnectionReport

type ConnectionReport struct {
	Reused           int
	Idle             int
	ConnectionErrors int
	IdleStats        Stats
	ConnectionStats  Stats
	// contains filtered or unexported fields
}

type DNSDone

type DNSDone struct {
	Timestamp    time.Time
	HadError     bool
	Coalesced    bool
	AddressCount int
}

type DNSReport

type DNSReport struct {
	Calls        int
	Errors       int
	Coalesced    int
	AddressCount int
	// All measurements in Milliseconds
	LatencyStats Stats
	// contains filtered or unexported fields
}

type DNSStart

type DNSStart struct {
	Timestamp time.Time
	// TODO: publish host-specific DNS details
	Host string
}

type GotConnection

type GotConnection struct {
	Timestamp time.Time
	Reused    bool
	WasIdle   bool
	IdleTime  time.Duration
}

type Key

type Key struct {
	Node  node.ID
	Chain types.RelayChainID
}

func (Key) String

func (k Key) String() string

type Limiter

type Limiter interface {
	Take(timestamp time.Time, tokens int64)
	Allow(timestamp time.Time, tokens int64) bool
}

func NewTokenBucket

func NewTokenBucket(rate int64, maxTokens int64) Limiter

type NetworkReport

type NetworkReport struct {
	DNSReport        `json:"dns,omitempty"`
	ConnectionReport `json:"connection,omitempty"`
	TLSReport        `json:"tls,omitempty"`
	ServerReport     `json:"node,omitempty"`
	RoundTripReport  `json:"roundtrip,omitempty"`
}

func (*NetworkReport) String

func (n *NetworkReport) String() string

type NetworkTracer

type NetworkTracer interface {
	protocol.NetworkTracer
	Start()
}

func NewNetworkTracer

func NewNetworkTracer(config NetworkTracerConfig, logger *logger.Logger) NetworkTracer

type NetworkTracerConfig

type NetworkTracerConfig struct {
	Workers                int
	MaxMeasurements        int
	Adjustments            int
	PublishIntervalSeconds int
}

func (NetworkTracerConfig) IsEmpty

func (n NetworkTracerConfig) IsEmpty() bool

type NewLimiter

type NewLimiter func(rate int64, maxTokens int64) Limiter

type PortalAppGetter

type PortalAppGetter interface {
	GetPortalAppByID(portalAppID types.PortalAppID) (types.PortalAppLite, error)
}

type Recorder

type Recorder interface {
	RecordRelay(relay Relay) error
	RecordSample(sample Sample) error
	GetSessionOutputChannel() chan<- provider.Session
}

func NewMetricsRecorder

func NewMetricsRecorder(nc, gnc *nats.Conn, config RecorderConfig, samplesChan chan Sample, metrics exporter.MetricExporter, portalRegion string, logger *logger.Logger) (Recorder, error)

TODO: REFACTOR: remove dependency on nats package

type RecorderConfig

type RecorderConfig struct {
	RelaysChanSize                 int `json:"relays_chan_size"`
	SamplesChanSize                int `json:"samples_chan_size"`
	WorkerPoolSize                 int `json:"worker_pool_size"`
	SessionPublishers              int `json:"session_publishers"`
	PercentageReportedToGlobalNats int `json:"percentage_reported_to_global_nats"`
}

func (*RecorderConfig) SetDefaults

func (r *RecorderConfig) SetDefaults()

type Relay

type Relay struct {
	RequestID string `json:"requestID"`

	PoktChainID       types.RelayChainID          `json:"poktChainID"`
	SessionKey        string                      `json:"sessionKey"`
	ProtocolPublicKey types.GigastakeAppPublicKey `json:"protocolPublicKey"`
	PoktNodePublicKey string                      `json:"poktNodePublicKey"`

	// TODO: change field name to Host: the URL contains the App ID, which is captured separately
	RelaySourceURL string `json:"relaySourceUrl"`

	// TODO: need better field names
	RelayStartDatetime  time.Time `json:"relayStartDatetime"`
	RelayReturnDatetime time.Time `json:"relayReturnDatetime"`

	NodeSendTime        time.Time `json:"nodeSendTime"`
	NodeReceiveTime     time.Time `json:"nodeReceiveTime"`
	AltruistSendTime    time.Time `json:"altruistSendTime"`
	AltruistReceiveTime time.Time `json:"altruistReceiveTime"`

	RelayRequest RelayRequest   `json:"relayRequest"`
	RelayMethods []types.Method `json:"relayMethods"`
	RelaySize    int            `json:"relaySize"`

	// Note: if the relay eventually succeeds, this will be false,
	//	but the ErrorType/Subtype will still provide data on any failures.
	//	an example is when node response fails validation, and altruist is used successfully.
	IsError      bool               `json:"isError"`
	ErrorCode    int                `json:"errorCode,omitempty"`
	ErrorMessage string             `json:"errorMessage,omitempty"`
	ErrorType    relay.ErrorType    `json:"errorType,omitempty"`
	ErrorSubtype relay.ErrorSubtype `json:"errorSubtype,omitempty"`
	// TODO: use error types instead
	BlockingPlugin string `json:"blockingPlugin"`

	RelayURLIsPublicEndpoint bool `json:"relayUrlIsPublicEndpoint"`

	IsAltruistRelay  bool            `json:"isAltruistRelay"`
	IsUserRelay      bool            `json:"isUserRelay"`
	PortalRegionName string          `json:"portalRegionName"`
	AccountID        types.AccountID `json:"accountID"`
	PoktNodeDomain   string          `json:"poktNodeDomain"`
	PoktNodeAddress  string          `json:"poktNodeAddress"`
	PoktNodeURL      string          `json:"poktNodeURL"`
	SessionHeight    int             `json:"sessionHeight,omitempty"`

	PoktTxID string `json:"poktTxID"`
}

func (*Relay) LoadFromLog

func (meta *Relay) LoadFromLog(r relay.RelayLog)

func (*Relay) String

func (meta *Relay) String() string

type RelayRequest

type RelayRequest struct {
	Details   relay.RelayDetails
	Origin    types.Origin
	UserAgent types.UserAgent
}

type RelayResult

type RelayResult int

const (
	Success RelayResult = 0
	Failure RelayResult = 1
)

type RoundTripReport

type RoundTripReport struct {
	RoundTripStats Stats
	// contains filtered or unexported fields
}

type Sample

type Sample struct {
	Key
	RequestID string
	Latency   float64
	Result    RelayResult
}

func GetSampleFromRelay

func GetSampleFromRelay(relay Relay) Sample

func (*Sample) String

func (s *Sample) String() string

type ServerReport

type ServerReport struct {
	LatencyStats Stats
	// contains filtered or unexported fields
}

type Stats

type Stats struct {
	Median float64
	Min    float64
	Max    float64
	Mean   float64
	P90    float64
	P95    float64
}

type TLSReport

type TLSReport struct {
	HandshakeStats Stats
	// contains filtered or unexported fields
}

type ThroughputAllower

type ThroughputAllower interface {
	IsRelayAllowed(appId string, limit int32, count int) bool
}

type ThroughputLimitConfig

type ThroughputLimitConfig struct {
	// How often throughput reports are sent to pubsub
	SendInterval int `json:"send_interval"`

	// Default free tier rate interval should it not be present in the db
	FreeTierRate int `json:"free_tier_rate"`

	// Minimum rate a free tier app must maintain to be tracked in prometheus
	RateMetricMinimum float64 `json:"rate_metric_minimum"`

	// RateTrackerHistorySize is how many seconds the rate tracker uses to calculate rate
	RateTrackerHistorySize int `json:"rate_tracker_history_size"`
}

type ThroughputMetric

type ThroughputMetric map[string]int

type ThroughputReport

type ThroughputReport struct {
	Hostname    string
	SendTime    time.Time
	Throughputs ThroughputMetric
}

type ThroughputStore

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

func NewThroughputStore

func NewThroughputStore(sendChan chan ThroughputReport, recvChan chan ThroughputReport, config ThroughputLimitConfig, newLimiter NewLimiter, rateTracker ratetracker.RateTracker, backend PortalAppGetter, metrics exporter.MetricExporter, logger *logger.Logger) *ThroughputStore

func (*ThroughputStore) IsRelayAllowed

func (t *ThroughputStore) IsRelayAllowed(appId string, limit int32, count int) (allowed bool)

IsRelayAllowed checks if a relay is allowed to be sent (is not rate-limited) This function also records the relay in localThroughputs

func (*ThroughputStore) Start

func (t *ThroughputStore) Start()

Start runs a goroutine that sends and receives throughput reports

type TokenBucket

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

func (*TokenBucket) Allow

func (t *TokenBucket) Allow(timestamp time.Time, count int64) bool

Allow returns whether count tokens can be consumed from the bucket. Tokens will be consumed if allowed.

func (*TokenBucket) Take

func (t *TokenBucket) Take(timestamp time.Time, tokens int64)

Take takes tokens from the bucket regardless of how many are remaining, allowing for token debt.