You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
617 lines
16 KiB
617 lines
16 KiB
// Copyright 2017 Tendermint. All rights reserved.
|
|
// Use of this source code is governed by Apache 2 LICENSE that can be found in the LICENSE file.
|
|
|
|
package trust
|
|
|
|
import (
|
|
"encoding/json"
|
|
"math"
|
|
"sync"
|
|
"time"
|
|
|
|
cmn "github.com/tendermint/tmlibs/common"
|
|
dbm "github.com/tendermint/tmlibs/db"
|
|
)
|
|
|
|
const defaultStorePeriodicSaveInterval = 1 * time.Minute
|
|
|
|
// TrustMetricStore - Manages all trust metrics for peers
|
|
type TrustMetricStore struct {
|
|
cmn.BaseService
|
|
|
|
// Maps a Peer.Key to that peer's TrustMetric
|
|
peerMetrics map[string]*TrustMetric
|
|
|
|
// Mutex that protects the map and history data file
|
|
mtx sync.Mutex
|
|
|
|
// The db where peer trust metric history data will be stored
|
|
db dbm.DB
|
|
|
|
// This configuration will be used when creating new TrustMetrics
|
|
config TrustMetricConfig
|
|
}
|
|
|
|
// NewTrustMetricStore returns a store that saves data to the DB
|
|
// and uses the config when creating new trust metrics
|
|
func NewTrustMetricStore(db dbm.DB, tmc TrustMetricConfig) *TrustMetricStore {
|
|
tms := &TrustMetricStore{
|
|
peerMetrics: make(map[string]*TrustMetric),
|
|
db: db,
|
|
config: tmc,
|
|
}
|
|
|
|
tms.BaseService = *cmn.NewBaseService(nil, "TrustMetricStore", tms)
|
|
return tms
|
|
}
|
|
|
|
// OnStart implements Service
|
|
func (tms *TrustMetricStore) OnStart() error {
|
|
if err := tms.BaseService.OnStart(); err != nil {
|
|
return err
|
|
}
|
|
|
|
tms.mtx.Lock()
|
|
defer tms.mtx.Unlock()
|
|
|
|
tms.loadFromDB()
|
|
go tms.saveRoutine()
|
|
return nil
|
|
}
|
|
|
|
// OnStop implements Service
|
|
func (tms *TrustMetricStore) OnStop() {
|
|
tms.BaseService.OnStop()
|
|
|
|
tms.mtx.Lock()
|
|
defer tms.mtx.Unlock()
|
|
|
|
// Stop all trust metric go-routines
|
|
for _, tm := range tms.peerMetrics {
|
|
tm.Stop()
|
|
}
|
|
|
|
// Make the final trust history data save
|
|
tms.saveToDB()
|
|
}
|
|
|
|
// Size returns the number of entries in the trust metric store
|
|
func (tms *TrustMetricStore) Size() int {
|
|
tms.mtx.Lock()
|
|
defer tms.mtx.Unlock()
|
|
|
|
return tms.size()
|
|
}
|
|
|
|
// GetPeerTrustMetric returns a trust metric by peer key
|
|
func (tms *TrustMetricStore) GetPeerTrustMetric(key string) *TrustMetric {
|
|
tms.mtx.Lock()
|
|
defer tms.mtx.Unlock()
|
|
|
|
tm, ok := tms.peerMetrics[key]
|
|
if !ok {
|
|
// If the metric is not available, we will create it
|
|
tm = NewMetricWithConfig(tms.config)
|
|
// The metric needs to be in the map
|
|
tms.peerMetrics[key] = tm
|
|
}
|
|
return tm
|
|
}
|
|
|
|
// PeerDisconnected pauses the trust metric associated with the peer identified by the key
|
|
func (tms *TrustMetricStore) PeerDisconnected(key string) {
|
|
tms.mtx.Lock()
|
|
defer tms.mtx.Unlock()
|
|
|
|
// If the Peer that disconnected has a metric, pause it
|
|
if tm, ok := tms.peerMetrics[key]; ok {
|
|
tm.Pause()
|
|
}
|
|
}
|
|
|
|
// Saves the history data for all peers to the store DB.
|
|
// This public method acquires the trust metric store lock
|
|
func (tms *TrustMetricStore) SaveToDB() {
|
|
tms.mtx.Lock()
|
|
defer tms.mtx.Unlock()
|
|
|
|
tms.saveToDB()
|
|
}
|
|
|
|
/* Private methods */
|
|
|
|
// size returns the number of entries in the store without acquiring the mutex
|
|
func (tms *TrustMetricStore) size() int {
|
|
return len(tms.peerMetrics)
|
|
}
|
|
|
|
/* Loading & Saving */
|
|
/* Both loadFromDB and savetoDB assume the mutex has been acquired */
|
|
|
|
var trustMetricKey = []byte("trustMetricStore")
|
|
|
|
// Loads the history data for all peers from the store DB
|
|
// cmn.Panics if file is corrupt
|
|
func (tms *TrustMetricStore) loadFromDB() bool {
|
|
// Obtain the history data we have so far
|
|
bytes := tms.db.Get(trustMetricKey)
|
|
if bytes == nil {
|
|
return false
|
|
}
|
|
|
|
peers := make(map[string]MetricHistoryJSON, 0)
|
|
err := json.Unmarshal(bytes, &peers)
|
|
if err != nil {
|
|
cmn.PanicCrisis(cmn.Fmt("Could not unmarshal Trust Metric Store DB data: %v", err))
|
|
}
|
|
|
|
// If history data exists in the file,
|
|
// load it into trust metric
|
|
for key, p := range peers {
|
|
tm := NewMetricWithConfig(tms.config)
|
|
|
|
tm.Init(p)
|
|
// Load the peer trust metric into the store
|
|
tms.peerMetrics[key] = tm
|
|
}
|
|
return true
|
|
}
|
|
|
|
// Saves the history data for all peers to the store DB
|
|
func (tms *TrustMetricStore) saveToDB() {
|
|
tms.Logger.Debug("Saving TrustHistory to DB", "size", tms.size())
|
|
|
|
peers := make(map[string]MetricHistoryJSON, 0)
|
|
|
|
for key, tm := range tms.peerMetrics {
|
|
// Add an entry for the peer identified by key
|
|
peers[key] = tm.HistoryJSON()
|
|
}
|
|
|
|
// Write all the data back to the DB
|
|
bytes, err := json.Marshal(peers)
|
|
if err != nil {
|
|
tms.Logger.Error("Failed to encode the TrustHistory", "err", err)
|
|
return
|
|
}
|
|
tms.db.SetSync(trustMetricKey, bytes)
|
|
}
|
|
|
|
// Periodically saves the trust history data to the DB
|
|
func (tms *TrustMetricStore) saveRoutine() {
|
|
t := time.NewTicker(defaultStorePeriodicSaveInterval)
|
|
defer t.Stop()
|
|
loop:
|
|
for {
|
|
select {
|
|
case <-t.C:
|
|
tms.SaveToDB()
|
|
case <-tms.Quit:
|
|
break loop
|
|
}
|
|
}
|
|
}
|
|
|
|
//---------------------------------------------------------------------------------------
|
|
|
|
const (
|
|
// The weight applied to the derivative when current behavior is >= previous behavior
|
|
defaultDerivativeGamma1 = 0
|
|
|
|
// The weight applied to the derivative when current behavior is less than previous behavior
|
|
defaultDerivativeGamma2 = 1.0
|
|
|
|
// The weight applied to history data values when calculating the history value
|
|
defaultHistoryDataWeight = 0.8
|
|
)
|
|
|
|
// TrustMetric - keeps track of peer reliability
|
|
// See tendermint/docs/architecture/adr-006-trust-metric.md for details
|
|
type TrustMetric struct {
|
|
// Mutex that protects the metric from concurrent access
|
|
mtx sync.Mutex
|
|
|
|
// Determines the percentage given to current behavior
|
|
proportionalWeight float64
|
|
|
|
// Determines the percentage given to prior behavior
|
|
integralWeight float64
|
|
|
|
// Count of how many time intervals this metric has been tracking
|
|
numIntervals int
|
|
|
|
// Size of the time interval window for this trust metric
|
|
maxIntervals int
|
|
|
|
// The time duration for a single time interval
|
|
intervalLen time.Duration
|
|
|
|
// Stores the trust history data for this metric
|
|
history []float64
|
|
|
|
// Weights applied to the history data when calculating the history value
|
|
historyWeights []float64
|
|
|
|
// The sum of the history weights used when calculating the history value
|
|
historyWeightSum float64
|
|
|
|
// The current number of history data elements
|
|
historySize int
|
|
|
|
// The maximum number of history data elements
|
|
historyMaxSize int
|
|
|
|
// The calculated history value for the current time interval
|
|
historyValue float64
|
|
|
|
// The number of recorded good and bad events for the current time interval
|
|
bad, good float64
|
|
|
|
// While true, history data is not modified
|
|
paused bool
|
|
|
|
// Signal channel for stopping the trust metric go-routine
|
|
stop chan struct{}
|
|
}
|
|
|
|
// MetricHistoryJSON - history data necessary to save the trust metric
|
|
type MetricHistoryJSON struct {
|
|
NumIntervals int `json:"intervals"`
|
|
History []float64 `json:"history"`
|
|
}
|
|
|
|
// Returns a snapshot of the trust metric history data
|
|
func (tm *TrustMetric) HistoryJSON() MetricHistoryJSON {
|
|
tm.mtx.Lock()
|
|
defer tm.mtx.Unlock()
|
|
|
|
return MetricHistoryJSON{
|
|
NumIntervals: tm.numIntervals,
|
|
History: tm.history,
|
|
}
|
|
}
|
|
|
|
// Instantiates a trust metric by loading the history data for a single peer.
|
|
// This is called only once and only right after creation, which is why the
|
|
// lock is not held while accessing the trust metric struct members
|
|
func (tm *TrustMetric) Init(hist MetricHistoryJSON) {
|
|
// Restore the number of time intervals we have previously tracked
|
|
if hist.NumIntervals > tm.maxIntervals {
|
|
hist.NumIntervals = tm.maxIntervals
|
|
}
|
|
tm.numIntervals = hist.NumIntervals
|
|
// Restore the history and its current size
|
|
if len(hist.History) > tm.historyMaxSize {
|
|
// Keep the history no larger than historyMaxSize
|
|
last := len(hist.History) - tm.historyMaxSize
|
|
hist.History = hist.History[last:]
|
|
}
|
|
tm.history = hist.History
|
|
tm.historySize = len(tm.history)
|
|
// Create the history weight values and weight sum
|
|
for i := 1; i <= tm.numIntervals; i++ {
|
|
x := math.Pow(defaultHistoryDataWeight, float64(i)) // Optimistic weight
|
|
tm.historyWeights = append(tm.historyWeights, x)
|
|
}
|
|
|
|
for _, v := range tm.historyWeights {
|
|
tm.historyWeightSum += v
|
|
}
|
|
// Calculate the history value based on the loaded history data
|
|
tm.historyValue = tm.calcHistoryValue()
|
|
}
|
|
|
|
// Pause tells the metric to pause recording data over time intervals.
|
|
// All method calls that indicate events will unpause the metric
|
|
func (tm *TrustMetric) Pause() {
|
|
tm.mtx.Lock()
|
|
defer tm.mtx.Unlock()
|
|
|
|
// Pause the metric for now
|
|
tm.paused = true
|
|
}
|
|
|
|
// Stop tells the metric to stop recording data over time intervals
|
|
func (tm *TrustMetric) Stop() {
|
|
tm.stop <- struct{}{}
|
|
}
|
|
|
|
// BadEvents indicates that an undesirable event(s) took place
|
|
func (tm *TrustMetric) BadEvents(num int) {
|
|
tm.mtx.Lock()
|
|
defer tm.mtx.Unlock()
|
|
|
|
tm.unpause()
|
|
tm.bad += float64(num)
|
|
}
|
|
|
|
// GoodEvents indicates that a desirable event(s) took place
|
|
func (tm *TrustMetric) GoodEvents(num int) {
|
|
tm.mtx.Lock()
|
|
defer tm.mtx.Unlock()
|
|
|
|
tm.unpause()
|
|
tm.good += float64(num)
|
|
}
|
|
|
|
// TrustValue gets the dependable trust value; always between 0 and 1
|
|
func (tm *TrustMetric) TrustValue() float64 {
|
|
tm.mtx.Lock()
|
|
defer tm.mtx.Unlock()
|
|
|
|
return tm.calcTrustValue()
|
|
}
|
|
|
|
// TrustScore gets a score based on the trust value always between 0 and 100
|
|
func (tm *TrustMetric) TrustScore() int {
|
|
score := tm.TrustValue() * 100
|
|
|
|
return int(math.Floor(score))
|
|
}
|
|
|
|
// NextTimeInterval saves current time interval data and prepares for the following interval
|
|
func (tm *TrustMetric) NextTimeInterval() {
|
|
tm.mtx.Lock()
|
|
defer tm.mtx.Unlock()
|
|
|
|
if tm.paused {
|
|
// Do not prepare for the next time interval while paused
|
|
return
|
|
}
|
|
|
|
// Add the current trust value to the history data
|
|
newHist := tm.calcTrustValue()
|
|
tm.history = append(tm.history, newHist)
|
|
|
|
// Update history and interval counters
|
|
if tm.historySize < tm.historyMaxSize {
|
|
tm.historySize++
|
|
} else {
|
|
// Keep the history no larger than historyMaxSize
|
|
last := len(tm.history) - tm.historyMaxSize
|
|
tm.history = tm.history[last:]
|
|
}
|
|
|
|
if tm.numIntervals < tm.maxIntervals {
|
|
tm.numIntervals++
|
|
// Add the optimistic weight for the new time interval
|
|
wk := math.Pow(defaultHistoryDataWeight, float64(tm.numIntervals))
|
|
tm.historyWeights = append(tm.historyWeights, wk)
|
|
tm.historyWeightSum += wk
|
|
}
|
|
|
|
// Update the history data using Faded Memories
|
|
tm.updateFadedMemory()
|
|
// Calculate the history value for the upcoming time interval
|
|
tm.historyValue = tm.calcHistoryValue()
|
|
tm.good = 0
|
|
tm.bad = 0
|
|
}
|
|
|
|
// Copy returns a new trust metric with members containing the same values
|
|
func (tm *TrustMetric) Copy() *TrustMetric {
|
|
if tm == nil {
|
|
return nil
|
|
}
|
|
|
|
return &TrustMetric{
|
|
proportionalWeight: tm.proportionalWeight,
|
|
integralWeight: tm.integralWeight,
|
|
numIntervals: tm.numIntervals,
|
|
maxIntervals: tm.maxIntervals,
|
|
intervalLen: tm.intervalLen,
|
|
history: tm.history,
|
|
historyWeights: tm.historyWeights,
|
|
historyWeightSum: tm.historyWeightSum,
|
|
historySize: tm.historySize,
|
|
historyMaxSize: tm.historyMaxSize,
|
|
historyValue: tm.historyValue,
|
|
good: tm.good,
|
|
bad: tm.bad,
|
|
paused: tm.paused,
|
|
stop: make(chan struct{}),
|
|
}
|
|
}
|
|
|
|
// TrustMetricConfig - Configures the weight functions and time intervals for the metric
|
|
type TrustMetricConfig struct {
|
|
// Determines the percentage given to current behavior
|
|
ProportionalWeight float64
|
|
|
|
// Determines the percentage given to prior behavior
|
|
IntegralWeight float64
|
|
|
|
// The window of time that the trust metric will track events across.
|
|
// This can be set to cover many days without issue
|
|
TrackingWindow time.Duration
|
|
|
|
// Each interval should be short for adapability.
|
|
// Less than 30 seconds is too sensitive,
|
|
// and greater than 5 minutes will make the metric numb
|
|
IntervalLength time.Duration
|
|
}
|
|
|
|
// DefaultConfig returns a config with values that have been tested and produce desirable results
|
|
func DefaultConfig() TrustMetricConfig {
|
|
return TrustMetricConfig{
|
|
ProportionalWeight: 0.4,
|
|
IntegralWeight: 0.6,
|
|
TrackingWindow: (time.Minute * 60 * 24) * 14, // 14 days.
|
|
IntervalLength: 1 * time.Minute,
|
|
}
|
|
}
|
|
|
|
// NewMetric returns a trust metric with the default configuration
|
|
func NewMetric() *TrustMetric {
|
|
return NewMetricWithConfig(DefaultConfig())
|
|
}
|
|
|
|
// NewMetricWithConfig returns a trust metric with a custom configuration
|
|
func NewMetricWithConfig(tmc TrustMetricConfig) *TrustMetric {
|
|
tm := new(TrustMetric)
|
|
config := customConfig(tmc)
|
|
|
|
// Setup using the configuration values
|
|
tm.proportionalWeight = config.ProportionalWeight
|
|
tm.integralWeight = config.IntegralWeight
|
|
tm.intervalLen = config.IntervalLength
|
|
// The maximum number of time intervals is the tracking window / interval length
|
|
tm.maxIntervals = int(config.TrackingWindow / tm.intervalLen)
|
|
// The history size will be determined by the maximum number of time intervals
|
|
tm.historyMaxSize = intervalToHistoryOffset(tm.maxIntervals) + 1
|
|
// This metric has a perfect history so far
|
|
tm.historyValue = 1.0
|
|
// Setup the stop channel
|
|
tm.stop = make(chan struct{})
|
|
|
|
go tm.processRequests()
|
|
return tm
|
|
}
|
|
|
|
/* Private methods */
|
|
|
|
// Ensures that all configuration elements have valid values
|
|
func customConfig(tmc TrustMetricConfig) TrustMetricConfig {
|
|
config := DefaultConfig()
|
|
|
|
// Check the config for set values, and setup appropriately
|
|
if tmc.ProportionalWeight > 0 {
|
|
config.ProportionalWeight = tmc.ProportionalWeight
|
|
}
|
|
|
|
if tmc.IntegralWeight > 0 {
|
|
config.IntegralWeight = tmc.IntegralWeight
|
|
}
|
|
|
|
if tmc.IntervalLength > time.Duration(0) {
|
|
config.IntervalLength = tmc.IntervalLength
|
|
}
|
|
|
|
if tmc.TrackingWindow > time.Duration(0) &&
|
|
tmc.TrackingWindow >= config.IntervalLength {
|
|
config.TrackingWindow = tmc.TrackingWindow
|
|
}
|
|
|
|
return config
|
|
}
|
|
|
|
// Wakes the trust metric up if it is currently paused
|
|
// This method needs to be called with the mutex locked
|
|
func (tm *TrustMetric) unpause() {
|
|
// Check if this is the first experience with
|
|
// what we are tracking since being paused
|
|
if tm.paused {
|
|
tm.good = 0
|
|
tm.bad = 0
|
|
// New events cause us to unpause the metric
|
|
tm.paused = false
|
|
}
|
|
}
|
|
|
|
// Calculates the derivative component
|
|
func (tm *TrustMetric) derivativeValue() float64 {
|
|
return tm.proportionalValue() - tm.historyValue
|
|
}
|
|
|
|
// Strengthens the derivative component when the change is negative
|
|
func (tm *TrustMetric) weightedDerivative() float64 {
|
|
var weight float64 = defaultDerivativeGamma1
|
|
|
|
d := tm.derivativeValue()
|
|
if d < 0 {
|
|
weight = defaultDerivativeGamma2
|
|
}
|
|
return weight * d
|
|
}
|
|
|
|
// Performs the update for our Faded Memories process, which allows the
|
|
// trust metric tracking window to be large while maintaining a small
|
|
// number of history data values
|
|
func (tm *TrustMetric) updateFadedMemory() {
|
|
if tm.historySize < 2 {
|
|
return
|
|
}
|
|
|
|
end := tm.historySize - 1
|
|
// Keep the most recent history element
|
|
for count := 1; count < tm.historySize; count++ {
|
|
i := end - count
|
|
// The older the data is, the more we spread it out
|
|
x := math.Pow(2, float64(count))
|
|
// Two history data values are merged into a single value
|
|
tm.history[i] = ((tm.history[i] * (x - 1)) + tm.history[i+1]) / x
|
|
}
|
|
}
|
|
|
|
// Map the interval value down to an offset from the beginning of history
|
|
func intervalToHistoryOffset(interval int) int {
|
|
// The system maintains 2^m interval values in the form of m history
|
|
// data values. Therefore, we access the ith interval by obtaining
|
|
// the history data index = the floor of log2(i)
|
|
return int(math.Floor(math.Log2(float64(interval))))
|
|
}
|
|
|
|
// Retrieves the actual history data value that represents the requested time interval
|
|
func (tm *TrustMetric) fadedMemoryValue(interval int) float64 {
|
|
first := tm.historySize - 1
|
|
|
|
if interval == 0 {
|
|
// Base case
|
|
return tm.history[first]
|
|
}
|
|
|
|
offset := intervalToHistoryOffset(interval)
|
|
return tm.history[first-offset]
|
|
}
|
|
|
|
// Calculates the integral (history) component of the trust value
|
|
func (tm *TrustMetric) calcHistoryValue() float64 {
|
|
var hv float64
|
|
|
|
for i := 0; i < tm.numIntervals; i++ {
|
|
hv += tm.fadedMemoryValue(i) * tm.historyWeights[i]
|
|
}
|
|
|
|
return hv / tm.historyWeightSum
|
|
}
|
|
|
|
// Calculates the current score for good/bad experiences
|
|
func (tm *TrustMetric) proportionalValue() float64 {
|
|
value := 1.0
|
|
|
|
total := tm.good + tm.bad
|
|
if total > 0 {
|
|
value = tm.good / total
|
|
}
|
|
return value
|
|
}
|
|
|
|
// Calculates the trust value for the request processing
|
|
func (tm *TrustMetric) calcTrustValue() float64 {
|
|
weightedP := tm.proportionalWeight * tm.proportionalValue()
|
|
weightedI := tm.integralWeight * tm.historyValue
|
|
weightedD := tm.weightedDerivative()
|
|
|
|
tv := weightedP + weightedI + weightedD
|
|
// Do not return a negative value.
|
|
if tv < 0 {
|
|
tv = 0
|
|
}
|
|
return tv
|
|
}
|
|
|
|
// This method is for a goroutine that handles all requests on the metric
|
|
func (tm *TrustMetric) processRequests() {
|
|
t := time.NewTicker(tm.intervalLen)
|
|
defer t.Stop()
|
|
loop:
|
|
for {
|
|
select {
|
|
case <-t.C:
|
|
tm.NextTimeInterval()
|
|
case <-tm.stop:
|
|
// Stop all further tracking for this metric
|
|
break loop
|
|
}
|
|
}
|
|
}
|