// 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 { tms.BaseService.OnStart() 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 } } }