You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@trafficcontrol.apache.org by ne...@apache.org on 2017/01/25 17:30:12 UTC
[01/20] incubator-trafficcontrol git commit: Fix TM2 DsStats
disabledLocations
Repository: incubator-trafficcontrol
Updated Branches:
refs/heads/master cef0859e0 -> 246e2e633
Fix TM2 DsStats disabledLocations
Project: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/commit/091b3c9a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/tree/091b3c9a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/diff/091b3c9a
Branch: refs/heads/master
Commit: 091b3c9ad48c706d9b16bf109e72fd74254fca80
Parents: ae137e5
Author: Robert Butts <ro...@gmail.com>
Authored: Tue Jan 24 15:18:31 2017 -0700
Committer: Dave Neuman <ne...@apache.org>
Committed: Wed Jan 25 10:29:46 2017 -0700
----------------------------------------------------------------------
.../traffic_monitor/deliveryservice/stat.go | 26 +++++++++++++++++---
.../traffic_monitor/deliveryservicedata/stat.go | 13 +++++++---
2 files changed, 31 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/091b3c9a/traffic_monitor/experimental/traffic_monitor/deliveryservice/stat.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/deliveryservice/stat.go b/traffic_monitor/experimental/traffic_monitor/deliveryservice/stat.go
index 1958f05..b377a26 100644
--- a/traffic_monitor/experimental/traffic_monitor/deliveryservice/stat.go
+++ b/traffic_monitor/experimental/traffic_monitor/deliveryservice/stat.go
@@ -131,6 +131,23 @@ func addAvailableData(dsStats Stats, crStates peer.Crstates, serverCachegroups m
dsStats.DeliveryService[deliveryService] = stat // TODO Necessary? Remove?
}
}
+
+ // TODO move to its own func?
+ for dsName, ds := range crStates.Deliveryservice {
+ stat, ok := dsStats.DeliveryService[dsName]
+ if !ok {
+ log.Warnf("CreateStats not adding disabledLocations for '%s': not found in Stats\n", dsName)
+ continue // TODO log warning? Error?
+ }
+
+ // TODO determine if a deep copy is necessary
+ stat.CommonStats.CachesDisabled = make([]string, len(ds.DisabledLocations), len(ds.DisabledLocations))
+ for i, v := range ds.DisabledLocations {
+ stat.CommonStats.CachesDisabled[i] = string(v)
+ }
+ dsStats.DeliveryService[dsName] = stat // TODO Necessary? Remove?
+ }
+
return dsStats, nil
}
@@ -474,18 +491,19 @@ func addStatCacheStats(s *dsdata.StatsOld, c dsdata.StatCacheStats, deliveryServ
}
func addCommonData(s *dsdata.StatsOld, c *dsdata.StatCommon, deliveryService enum.DeliveryServiceName, t int64, filter dsdata.Filter) *dsdata.StatsOld {
- add := func(name, val string) {
+ add := func(name string, val interface{}) {
if filter.UseStat(name) {
s.DeliveryService[deliveryService][dsdata.StatName(name)] = []dsdata.StatOld{dsdata.StatOld{Time: t, Value: val}}
}
}
- add("caches-configured", strconv.Itoa(int(c.CachesConfiguredNum.Value)))
- add("caches-reporting", strconv.Itoa(len(c.CachesReporting)))
+ add("caches-configured", fmt.Sprintf("%d", c.CachesConfiguredNum.Value))
+ add("caches-reporting", fmt.Sprintf("%d", len(c.CachesReporting)))
add("error-string", c.ErrorStr.Value)
add("status", c.StatusStr.Value)
add("isHealthy", fmt.Sprintf("%t", c.IsHealthy.Value))
add("isAvailable", fmt.Sprintf("%t", c.IsAvailable.Value))
- add("caches-available", strconv.Itoa(int(c.CachesAvailableNum.Value)))
+ add("caches-available", fmt.Sprintf("%d", c.CachesAvailableNum.Value))
+ add("disabledLocations", c.CachesDisabled)
return s
}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/091b3c9a/traffic_monitor/experimental/traffic_monitor/deliveryservicedata/stat.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/deliveryservicedata/stat.go b/traffic_monitor/experimental/traffic_monitor/deliveryservicedata/stat.go
index ae79b08..b5d54b1 100644
--- a/traffic_monitor/experimental/traffic_monitor/deliveryservicedata/stat.go
+++ b/traffic_monitor/experimental/traffic_monitor/deliveryservicedata/stat.go
@@ -41,10 +41,10 @@ type StatName string
// StatOld is the old JSON representation of a stat, from Traffic Monitor 1.0.
type StatOld struct {
- Time int64 `json:"time"`
- Value string `json:"value"`
- Span int `json:"span,omitempty"` // TODO set? remove?
- Index int `json:"index,omitempty"` // TODO set? remove?
+ Time int64 `json:"time"`
+ Value interface{} `json:"value"`
+ Span int `json:"span,omitempty"` // TODO set? remove?
+ Index int `json:"index,omitempty"` // TODO set? remove?
}
// StatsOld is the old JSON representation of stats, from Traffic Monitor 1.0. It is designed to be serialized and returns from an API, and includes stat history for each delivery service, as well as data common to most endpoints.
@@ -118,6 +118,7 @@ type StatCommon struct {
IsHealthy StatBool `json:"is_healthy"`
IsAvailable StatBool `json:"is_available"`
CachesAvailableNum StatInt `json:"caches_available"`
+ CachesDisabled []string `json:"disabled_locations"`
}
// Copy returns a deep copy of this StatCommon object.
@@ -126,6 +127,10 @@ func (a StatCommon) Copy() StatCommon {
for k, v := range a.CachesReporting {
b.CachesReporting[k] = v
}
+ b.CachesDisabled = make([]string, len(a.CachesDisabled), len(a.CachesDisabled))
+ for i, v := range a.CachesDisabled {
+ b.CachesDisabled[i] = v
+ }
return b
}
[19/20] incubator-trafficcontrol git commit: Move TM2
CalcAvailability to health package
Posted by ne...@apache.org.
Move TM2 CalcAvailability to health package
Project: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/commit/dea90b1e
Tree: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/tree/dea90b1e
Diff: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/diff/dea90b1e
Branch: refs/heads/master
Commit: dea90b1ebd8dd10e3080b6c238e476e817022b66
Parents: 7a748fe
Author: Robert Butts <ro...@gmail.com>
Authored: Tue Jan 24 11:52:36 2017 -0700
Committer: Dave Neuman <ne...@apache.org>
Committed: Wed Jan 25 10:29:46 2017 -0700
----------------------------------------------------------------------
.../traffic_monitor/health/cache.go | 79 +++++++++++++++++---
.../traffic_monitor/health/event.go | 47 ++++++++++++
.../traffic_monitor/manager/datarequest.go | 4 +-
.../traffic_monitor/manager/health.go | 29 +------
.../traffic_monitor/manager/manager.go | 3 +-
.../traffic_monitor/manager/opsconfig.go | 3 +-
.../traffic_monitor/manager/peer.go | 9 +--
.../traffic_monitor/manager/stat.go | 43 +----------
.../traffic_monitor/threadsafe/events.go | 71 ------------------
9 files changed, 134 insertions(+), 154 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/dea90b1e/traffic_monitor/experimental/traffic_monitor/health/cache.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/health/cache.go b/traffic_monitor/experimental/traffic_monitor/health/cache.go
index aec1d8f..533652a 100644
--- a/traffic_monitor/experimental/traffic_monitor/health/cache.go
+++ b/traffic_monitor/experimental/traffic_monitor/health/cache.go
@@ -23,19 +23,17 @@ import (
"fmt"
"strconv"
"strings"
+ "time"
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/common/log"
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/cache"
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/enum"
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/peer"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/threadsafe"
+ todata "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/trafficopsdata"
to "github.com/apache/incubator-trafficcontrol/traffic_ops/client"
)
-func setErr(newResult *cache.Result, err error) {
- newResult.Error = err
- newResult.Available = false
-}
-
// GetVitals Gets the vitals to decide health on in the right format
func GetVitals(newResult *cache.Result, prevResult *cache.Result, mc *to.TrafficMonitorConfigMap) {
if newResult.Error != nil {
@@ -161,16 +159,58 @@ func EvalCache(result cache.ResultInfo, resultStats cache.ResultStatValHistory,
continue
}
- if !InThreshold(threshold, resultStatNum) {
- return false, eventDesc(status, ExceedsThresholdMsg(stat, threshold, resultStatNum)), stat
+ if !inThreshold(threshold, resultStatNum) {
+ return false, eventDesc(status, exceedsThresholdMsg(stat, threshold, resultStatNum)), stat
}
}
return result.Available, eventDesc(status, availability), ""
}
+// CalcAvailability calculates the availability of the cache, from the given result. Availability is stored in `localCacheStatus` and `localStates`, and if the status changed an event is added to `events`. statResultHistory may be nil, for pollers which don't poll stats.
+// TODO add enum for poller names?
+func CalcAvailability(results []cache.Result, pollerName string, statResultHistory cache.ResultStatHistory, mc to.TrafficMonitorConfigMap, toData todata.TOData, localCacheStatusThreadsafe threadsafe.CacheAvailableStatus, localStates peer.CRStatesThreadsafe, events ThreadsafeEvents) {
+ localCacheStatuses := localCacheStatusThreadsafe.Get().Copy()
+ for _, result := range results {
+ statResults := cache.ResultStatValHistory(nil)
+ if statResultHistory != nil {
+ statResults = statResultHistory[result.ID]
+ }
+
+ isAvailable, whyAvailable, unavailableStat := EvalCache(cache.ToInfo(result), statResults, &mc)
+
+ // if the cache is now Available, and was previously unavailable due to a threshold, make sure this poller contains the stat which exceeded the threshold.
+ if previousStatus, hasPreviousStatus := localCacheStatuses[result.ID]; isAvailable && hasPreviousStatus && !previousStatus.Available && previousStatus.UnavailableStat != "" {
+ if !result.HasStat(previousStatus.UnavailableStat) {
+ return
+ }
+ }
+ localCacheStatuses[result.ID] = cache.AvailableStatus{
+ Available: isAvailable,
+ Status: mc.TrafficServer[string(result.ID)].Status,
+ Why: whyAvailable,
+ UnavailableStat: unavailableStat,
+ Poller: pollerName,
+ } // TODO move within localStates?
+
+ if available, ok := localStates.GetCache(result.ID); !ok || available.IsAvailable != isAvailable {
+ log.Infof("Changing state for %s was: %t now: %t because %s poller: %v error: %v", result.ID, available.IsAvailable, isAvailable, whyAvailable, pollerName, result.Error)
+ events.Add(Event{Time: time.Now(), Description: whyAvailable + " (" + pollerName + ")", Name: string(result.ID), Hostname: string(result.ID), Type: toData.ServerTypes[result.ID].String(), Available: isAvailable})
+ }
+
+ localStates.SetCache(result.ID, peer.IsAvailable{IsAvailable: isAvailable})
+ }
+ calculateDeliveryServiceState(toData.DeliveryServiceServers, localStates)
+ localCacheStatusThreadsafe.Set(localCacheStatuses)
+}
+
+func setErr(newResult *cache.Result, err error) {
+ newResult.Error = err
+ newResult.Available = false
+}
+
// ExceedsThresholdMsg returns a human-readable message for why the given value exceeds the threshold. It does NOT check whether the value actually exceeds the threshold; call `InThreshold` to check first.
-func ExceedsThresholdMsg(stat string, threshold to.HealthThreshold, val float64) string {
+func exceedsThresholdMsg(stat string, threshold to.HealthThreshold, val float64) string {
switch threshold.Comparator {
case "=":
return fmt.Sprintf("%s not equal (%f != %f)", stat, val, threshold.Val)
@@ -187,7 +227,7 @@ func ExceedsThresholdMsg(stat string, threshold to.HealthThreshold, val float64)
}
}
-func InThreshold(threshold to.HealthThreshold, val float64) bool {
+func inThreshold(threshold to.HealthThreshold, val float64) bool {
switch threshold.Comparator {
case "=":
return val == threshold.Val
@@ -208,3 +248,24 @@ func InThreshold(threshold to.HealthThreshold, val float64) bool {
func eventDesc(status enum.CacheStatus, message string) string {
return fmt.Sprintf("%s - %s", status, message)
}
+
+// calculateDeliveryServiceState calculates the state of delivery services from the new cache state data `cacheState` and the CRConfig data `deliveryServiceServers` and puts the calculated state in the outparam `deliveryServiceStates`
+func calculateDeliveryServiceState(deliveryServiceServers map[enum.DeliveryServiceName][]enum.CacheName, states peer.CRStatesThreadsafe) {
+ deliveryServices := states.GetDeliveryServices()
+ for deliveryServiceName, deliveryServiceState := range deliveryServices {
+ if _, ok := deliveryServiceServers[deliveryServiceName]; !ok {
+ // log.Errorf("CRConfig does not have delivery service %s, but traffic monitor poller does; skipping\n", deliveryServiceName)
+ continue
+ }
+ deliveryServiceState.IsAvailable = false
+ deliveryServiceState.DisabledLocations = []enum.CacheName{} // it's important this isn't nil, so it serialises to the JSON `[]` instead of `null`
+ for _, server := range deliveryServiceServers[deliveryServiceName] {
+ if available, _ := states.GetCache(server); available.IsAvailable {
+ deliveryServiceState.IsAvailable = true
+ } else {
+ deliveryServiceState.DisabledLocations = append(deliveryServiceState.DisabledLocations, server)
+ }
+ }
+ states.SetDeliveryService(deliveryServiceName, deliveryServiceState)
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/dea90b1e/traffic_monitor/experimental/traffic_monitor/health/event.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/health/event.go b/traffic_monitor/experimental/traffic_monitor/health/event.go
index 47563fa..3150da4 100644
--- a/traffic_monitor/experimental/traffic_monitor/health/event.go
+++ b/traffic_monitor/experimental/traffic_monitor/health/event.go
@@ -20,7 +20,10 @@ package health
*/
import (
+ "sync"
"time"
+
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/common/log"
)
// Event represents an event change in aggregated data. For example, a cache being marked as unavailable.
@@ -34,3 +37,47 @@ type Event struct {
Type string `json:"type"`
Available bool `json:"isAvailable"`
}
+
+// Events provides safe access for multiple goroutines readers and a single writer to a stored Events slice.
+type ThreadsafeEvents struct {
+ events *[]Event
+ m *sync.RWMutex
+ nextIndex *uint64
+ max uint64
+}
+
+func copyEvents(a []Event) []Event {
+ b := make([]Event, len(a), len(a))
+ copy(b, a)
+ return b
+}
+
+// NewEvents creates a new single-writer-multiple-reader Threadsafe object
+func NewThreadsafeEvents(maxEvents uint64) ThreadsafeEvents {
+ i := uint64(0)
+ return ThreadsafeEvents{m: &sync.RWMutex{}, events: &[]Event{}, nextIndex: &i, max: maxEvents}
+}
+
+// Get returns the internal slice of Events for reading. This MUST NOT be modified. If modification is necessary, copy the slice.
+func (o *ThreadsafeEvents) Get() []Event {
+ o.m.RLock()
+ defer o.m.RUnlock()
+ return *o.events
+}
+
+// Add adds the given event. This is threadsafe for one writer, multiple readers. This MUST NOT be called by multiple threads, as it non-atomically fetches and adds.
+func (o *ThreadsafeEvents) Add(e Event) {
+ // host="hostname", type=EDGE, available=true, msg="REPORTED - available"
+ log.Eventf(e.Time, "host=\"%s\", type=%s, available=%t, msg=\"%s\"", e.Hostname, e.Type, e.Available, e.Description)
+ o.m.Lock() // TODO test removing
+ events := copyEvents(*o.events)
+ e.Index = *o.nextIndex
+ events = append([]Event{e}, events...)
+ if len(events) > int(o.max) {
+ events = (events)[:o.max-1]
+ }
+ // o.m.Lock()
+ *o.events = events
+ *o.nextIndex++
+ o.m.Unlock()
+}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/dea90b1e/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go b/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
index 30fcc47..eddf972 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
@@ -592,7 +592,7 @@ func srvDSStats(params url.Values, errorCount threadsafe.Uint, path string, toDa
return WrapErrCode(errorCount, path, bytes, err)
}
-func srvEventLog(events threadsafe.Events) ([]byte, error) {
+func srvEventLog(events health.ThreadsafeEvents) ([]byte, error) {
return json.Marshal(JSONEvents{Events: events.Get()})
}
@@ -704,7 +704,7 @@ func MakeDispatchMap(
statMaxKbpses threadsafe.CacheKbpses,
healthHistory threadsafe.ResultHistory,
dsStats threadsafe.DSStatsReader,
- events threadsafe.Events,
+ events health.ThreadsafeEvents,
staticAppData StaticAppData,
healthPollInterval time.Duration,
lastHealthDurations DurationMapThreadsafe,
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/dea90b1e/traffic_monitor/experimental/traffic_monitor/manager/health.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/health.go b/traffic_monitor/experimental/traffic_monitor/manager/health.go
index 63248e8..e2657f8 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/health.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/health.go
@@ -86,7 +86,7 @@ func StartHealthResultManager(
fetchCount threadsafe.Uint,
errorCount threadsafe.Uint,
cfg config.Config,
- events threadsafe.Events,
+ events health.ThreadsafeEvents,
localCacheStatus threadsafe.CacheAvailableStatus,
) (DurationMapThreadsafe, threadsafe.ResultHistory) {
lastHealthDurations := NewDurationMapThreadsafe()
@@ -120,7 +120,7 @@ func healthResultManagerListen(
combinedStates peer.CRStatesThreadsafe,
fetchCount threadsafe.Uint,
errorCount threadsafe.Uint,
- events threadsafe.Events,
+ events health.ThreadsafeEvents,
localCacheStatus threadsafe.CacheAvailableStatus,
cfg config.Config,
) {
@@ -186,7 +186,7 @@ func processHealthResult(
combinedStates peer.CRStatesThreadsafe,
fetchCount threadsafe.Uint,
errorCount threadsafe.Uint,
- events threadsafe.Events,
+ events health.ThreadsafeEvents,
localCacheStatusThreadsafe threadsafe.CacheAvailableStatus,
lastHealthEndTimes map[enum.CacheName]time.Time,
healthHistory threadsafe.ResultHistory,
@@ -229,7 +229,7 @@ func processHealthResult(
healthHistoryCopy[healthResult.ID] = pruneHistory(append([]cache.Result{healthResult}, healthHistoryCopy[healthResult.ID]...), maxHistory)
}
- calcAvailability(results, "health", nil, monitorConfigCopy, toDataCopy, localCacheStatusThreadsafe, localStates, events)
+ health.CalcAvailability(results, "health", nil, monitorConfigCopy, toDataCopy, localCacheStatusThreadsafe, localStates, events)
healthHistory.Set(healthHistoryCopy)
// TODO determine if we should combineCrStates() here
@@ -244,24 +244,3 @@ func processHealthResult(
}
lastHealthDurationsThreadsafe.Set(lastHealthDurations)
}
-
-// calculateDeliveryServiceState calculates the state of delivery services from the new cache state data `cacheState` and the CRConfig data `deliveryServiceServers` and puts the calculated state in the outparam `deliveryServiceStates`
-func CalculateDeliveryServiceState(deliveryServiceServers map[enum.DeliveryServiceName][]enum.CacheName, states peer.CRStatesThreadsafe) {
- deliveryServices := states.GetDeliveryServices()
- for deliveryServiceName, deliveryServiceState := range deliveryServices {
- if _, ok := deliveryServiceServers[deliveryServiceName]; !ok {
- // log.Errorf("CRConfig does not have delivery service %s, but traffic monitor poller does; skipping\n", deliveryServiceName)
- continue
- }
- deliveryServiceState.IsAvailable = false
- deliveryServiceState.DisabledLocations = []enum.CacheName{} // it's important this isn't nil, so it serialises to the JSON `[]` instead of `null`
- for _, server := range deliveryServiceServers[deliveryServiceName] {
- if available, _ := states.GetCache(server); available.IsAvailable {
- deliveryServiceState.IsAvailable = true
- } else {
- deliveryServiceState.DisabledLocations = append(deliveryServiceState.DisabledLocations, server)
- }
- }
- states.SetDeliveryService(deliveryServiceName, deliveryServiceState)
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/dea90b1e/traffic_monitor/experimental/traffic_monitor/manager/manager.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/manager.go b/traffic_monitor/experimental/traffic_monitor/manager/manager.go
index 79c69af..974b3eb 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/manager.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/manager.go
@@ -29,6 +29,7 @@ import (
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/common/poller"
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/cache"
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/config"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/health"
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/peer"
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/threadsafe"
todata "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/trafficopsdata"
@@ -86,7 +87,7 @@ func Start(opsConfigFile string, cfg config.Config, staticAppData StaticAppData)
go cacheStatPoller.Poll()
go peerPoller.Poll()
- events := threadsafe.NewEvents(cfg.MaxEvents)
+ events := health.NewThreadsafeEvents(cfg.MaxEvents)
cachesChanged := make(chan struct{})
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/dea90b1e/traffic_monitor/experimental/traffic_monitor/manager/opsconfig.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/opsconfig.go b/traffic_monitor/experimental/traffic_monitor/manager/opsconfig.go
index 29ffd01..8f7a0d2 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/opsconfig.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/opsconfig.go
@@ -28,6 +28,7 @@ import (
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/common/log"
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/common/poller"
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/config"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/health"
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/peer"
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/srvhttp"
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/threadsafe"
@@ -79,7 +80,7 @@ func StartOpsConfigManager(
healthHistory threadsafe.ResultHistory,
lastStats threadsafe.LastStats,
dsStats threadsafe.DSStatsReader,
- events threadsafe.Events,
+ events health.ThreadsafeEvents,
staticAppData StaticAppData,
healthPollInterval time.Duration,
lastHealthDurations DurationMapThreadsafe,
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/dea90b1e/traffic_monitor/experimental/traffic_monitor/manager/peer.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/peer.go b/traffic_monitor/experimental/traffic_monitor/manager/peer.go
index 13b8411..3b4a209 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/peer.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/peer.go
@@ -31,7 +31,6 @@ import (
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/enum"
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/health"
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/peer"
- "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/threadsafe"
todata "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/trafficopsdata"
)
@@ -40,11 +39,11 @@ func StartPeerManager(
peerChan <-chan peer.Result,
localStates peer.CRStatesThreadsafe,
peerStates peer.CRStatesPeersThreadsafe,
- events threadsafe.Events,
+ events health.ThreadsafeEvents,
peerOptimistic bool,
toData todata.TODataThreadsafe,
cfg config.Config,
-) (peer.CRStatesThreadsafe, threadsafe.Events) {
+) (peer.CRStatesThreadsafe, health.ThreadsafeEvents) {
combinedStates := peer.NewCRStatesThreadsafe()
overrideMap := map[enum.CacheName]bool{}
@@ -59,14 +58,14 @@ func StartPeerManager(
return combinedStates, events
}
-func comparePeerState(events threadsafe.Events, result peer.Result, peerStates peer.CRStatesPeersThreadsafe) {
+func comparePeerState(events health.ThreadsafeEvents, result peer.Result, peerStates peer.CRStatesPeersThreadsafe) {
if result.Available != peerStates.GetPeerAvailability(result.ID) {
events.Add(health.Event{Time: result.Time, Unix: result.Time.Unix(), Description: util.JoinErrorsString(result.Errors), Name: result.ID.String(), Hostname: result.ID.String(), Type: "Peer", Available: result.Available})
}
}
// TODO JvD: add deliveryservice stuff
-func combineCrStates(events threadsafe.Events, peerOptimistic bool, peerStates peer.CRStatesPeersThreadsafe, localStates peer.Crstates, combinedStates peer.CRStatesThreadsafe, overrideMap map[enum.CacheName]bool, toData todata.TODataThreadsafe) {
+func combineCrStates(events health.ThreadsafeEvents, peerOptimistic bool, peerStates peer.CRStatesPeersThreadsafe, localStates peer.Crstates, combinedStates peer.CRStatesThreadsafe, overrideMap map[enum.CacheName]bool, toData todata.TODataThreadsafe) {
toDataCopy := toData.Get()
for cacheName, localCacheState := range localStates.Caches { // localStates gets pruned when servers are disabled, it's the source of truth
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/dea90b1e/traffic_monitor/experimental/traffic_monitor/manager/stat.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/stat.go b/traffic_monitor/experimental/traffic_monitor/manager/stat.go
index a4a9b81..8e69bda 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/stat.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/stat.go
@@ -67,7 +67,7 @@ func StartStatHistoryManager(
errorCount threadsafe.Uint,
cfg config.Config,
monitorConfig TrafficMonitorConfigMapThreadsafe,
- events threadsafe.Events,
+ events health.ThreadsafeEvents,
) (threadsafe.ResultInfoHistory, threadsafe.ResultStatHistory, threadsafe.CacheKbpses, DurationMapThreadsafe, threadsafe.LastStats, threadsafe.DSStatsReader, threadsafe.UnpolledCaches, threadsafe.CacheAvailableStatus) {
statInfoHistory := threadsafe.NewResultInfoHistory()
statResultHistory := threadsafe.NewResultStatHistory()
@@ -141,7 +141,7 @@ func processStatResults(
precomputedData map[enum.CacheName]cache.PrecomputedData,
lastResults map[enum.CacheName]cache.Result,
localStates peer.CRStatesThreadsafe,
- events threadsafe.Events,
+ events health.ThreadsafeEvents,
localCacheStatusThreadsafe threadsafe.CacheAvailableStatus,
) {
if len(results) == 0 {
@@ -212,7 +212,7 @@ func processStatResults(
lastStats.Set(newLastStats)
}
- calcAvailability(results, "stat", statResultHistory, mc, toData, localCacheStatusThreadsafe, localStates, events)
+ health.CalcAvailability(results, "stat", statResultHistory, mc, toData, localCacheStatusThreadsafe, localStates, events)
endTime := time.Now()
lastStatDurations := lastStatDurationsThreadsafe.Get().Copy()
@@ -226,40 +226,3 @@ func processStatResults(
lastStatDurationsThreadsafe.Set(lastStatDurations)
unpolledCaches.SetPolled(results, lastStats.Get())
}
-
-// calcAvailability calculates the availability of the cache, from the given result. Availability is stored in `localCacheStatus` and `localStates`, and if the status changed an event is added to `events`. statResultHistory may be nil, for pollers which don't poll stats.
-// TODO add enum for poller names?
-func calcAvailability(results []cache.Result, pollerName string, statResultHistory cache.ResultStatHistory, mc to.TrafficMonitorConfigMap, toData todata.TOData, localCacheStatusThreadsafe threadsafe.CacheAvailableStatus, localStates peer.CRStatesThreadsafe, events threadsafe.Events) {
- localCacheStatuses := localCacheStatusThreadsafe.Get().Copy()
- for _, result := range results {
- statResults := cache.ResultStatValHistory(nil)
- if statResultHistory != nil {
- statResults = statResultHistory[result.ID]
- }
-
- isAvailable, whyAvailable, unavailableStat := health.EvalCache(cache.ToInfo(result), statResults, &mc)
-
- // if the cache is now Available, and was previously unavailable due to a threshold, make sure this poller contains the stat which exceeded the threshold.
- if previousStatus, hasPreviousStatus := localCacheStatuses[result.ID]; isAvailable && hasPreviousStatus && !previousStatus.Available && previousStatus.UnavailableStat != "" {
- if !result.HasStat(previousStatus.UnavailableStat) {
- return
- }
- }
- localCacheStatuses[result.ID] = cache.AvailableStatus{
- Available: isAvailable,
- Status: mc.TrafficServer[string(result.ID)].Status,
- Why: whyAvailable,
- UnavailableStat: unavailableStat,
- Poller: pollerName,
- } // TODO move within localStates?
-
- if available, ok := localStates.GetCache(result.ID); !ok || available.IsAvailable != isAvailable {
- log.Infof("Changing state for %s was: %t now: %t because %s poller: %v error: %v", result.ID, available.IsAvailable, isAvailable, whyAvailable, pollerName, result.Error)
- events.Add(health.Event{Time: time.Now(), Description: whyAvailable + " (" + pollerName + ")", Name: string(result.ID), Hostname: string(result.ID), Type: toData.ServerTypes[result.ID].String(), Available: isAvailable})
- }
-
- localStates.SetCache(result.ID, peer.IsAvailable{IsAvailable: isAvailable})
- }
- CalculateDeliveryServiceState(toData.DeliveryServiceServers, localStates)
- localCacheStatusThreadsafe.Set(localCacheStatuses)
-}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/dea90b1e/traffic_monitor/experimental/traffic_monitor/threadsafe/events.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/threadsafe/events.go b/traffic_monitor/experimental/traffic_monitor/threadsafe/events.go
deleted file mode 100644
index 9cd0320..0000000
--- a/traffic_monitor/experimental/traffic_monitor/threadsafe/events.go
+++ /dev/null
@@ -1,71 +0,0 @@
-package threadsafe
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-import (
- "sync"
-
- "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/common/log"
- "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/health"
-)
-
-// Events provides safe access for multiple goroutines readers and a single writer to a stored Events slice.
-type Events struct {
- events *[]health.Event
- m *sync.RWMutex
- nextIndex *uint64
- max uint64
-}
-
-func copyEvents(a []health.Event) []health.Event {
- b := make([]health.Event, len(a), len(a))
- copy(b, a)
- return b
-}
-
-// NewEvents creates a new single-writer-multiple-reader Threadsafe object
-func NewEvents(maxEvents uint64) Events {
- i := uint64(0)
- return Events{m: &sync.RWMutex{}, events: &[]health.Event{}, nextIndex: &i, max: maxEvents}
-}
-
-// Get returns the internal slice of Events for reading. This MUST NOT be modified. If modification is necessary, copy the slice.
-func (o *Events) Get() []health.Event {
- o.m.RLock()
- defer o.m.RUnlock()
- return *o.events
-}
-
-// Add adds the given event. This is threadsafe for one writer, multiple readers. This MUST NOT be called by multiple threads, as it non-atomically fetches and adds.
-func (o *Events) Add(e health.Event) {
- // host="hostname", type=EDGE, available=true, msg="REPORTED - available"
- log.Eventf(e.Time, "host=\"%s\", type=%s, available=%t, msg=\"%s\"", e.Hostname, e.Type, e.Available, e.Description)
- o.m.Lock() // TODO test removing
- events := copyEvents(*o.events)
- e.Index = *o.nextIndex
- events = append([]health.Event{e}, events...)
- if len(events) > int(o.max) {
- events = (events)[:o.max-1]
- }
- // o.m.Lock()
- *o.events = events
- *o.nextIndex++
- o.m.Unlock()
-}
[10/20] incubator-trafficcontrol git commit: Rename TM2
getEventDescription to eventDesc
Posted by ne...@apache.org.
Rename TM2 getEventDescription to eventDesc
Project: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/commit/7a748fe3
Tree: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/tree/7a748fe3
Diff: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/diff/7a748fe3
Branch: refs/heads/master
Commit: 7a748fe3a22edd60529e8da94242fbd8f20e9261
Parents: 0d898cc
Author: Robert Butts <ro...@gmail.com>
Authored: Tue Jan 24 11:16:38 2017 -0700
Committer: Dave Neuman <ne...@apache.org>
Committed: Wed Jan 25 10:29:46 2017 -0700
----------------------------------------------------------------------
.../experimental/traffic_monitor/health/cache.go | 16 ++++++++--------
1 file changed, 8 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/7a748fe3/traffic_monitor/experimental/traffic_monitor/health/cache.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/health/cache.go b/traffic_monitor/experimental/traffic_monitor/health/cache.go
index 68f64a7..aec1d8f 100644
--- a/traffic_monitor/experimental/traffic_monitor/health/cache.go
+++ b/traffic_monitor/experimental/traffic_monitor/health/cache.go
@@ -122,16 +122,16 @@ func EvalCache(result cache.ResultInfo, resultStats cache.ResultStatValHistory,
switch {
case status == enum.CacheStatusInvalid:
log.Errorf("Cache %v got invalid status from Traffic Ops '%v' - treating as OFFLINE\n", result.ID, serverInfo.Status)
- return false, getEventDescription(status, availability+"; invalid status"), ""
+ return false, eventDesc(status, availability+"; invalid status"), ""
case status == enum.CacheStatusAdminDown:
- return false, getEventDescription(status, availability), ""
+ return false, eventDesc(status, availability), ""
case status == enum.CacheStatusOffline:
log.Errorf("Cache %v set to offline, but still polled\n", result.ID)
- return false, getEventDescription(status, availability), ""
+ return false, eventDesc(status, availability), ""
case status == enum.CacheStatusOnline:
- return true, getEventDescription(status, availability), ""
+ return true, eventDesc(status, availability), ""
case result.Error != nil:
- return false, getEventDescription(status, fmt.Sprintf("%v", result.Error)), ""
+ return false, eventDesc(status, fmt.Sprintf("%v", result.Error)), ""
}
computedStats := cache.ComputedStats()
@@ -162,11 +162,11 @@ func EvalCache(result cache.ResultInfo, resultStats cache.ResultStatValHistory,
}
if !InThreshold(threshold, resultStatNum) {
- return false, getEventDescription(status, ExceedsThresholdMsg(stat, threshold, resultStatNum)), stat
+ return false, eventDesc(status, ExceedsThresholdMsg(stat, threshold, resultStatNum)), stat
}
}
- return result.Available, getEventDescription(status, availability), ""
+ return result.Available, eventDesc(status, availability), ""
}
// ExceedsThresholdMsg returns a human-readable message for why the given value exceeds the threshold. It does NOT check whether the value actually exceeds the threshold; call `InThreshold` to check first.
@@ -205,6 +205,6 @@ func InThreshold(threshold to.HealthThreshold, val float64) bool {
}
}
-func getEventDescription(status enum.CacheStatus, message string) string {
+func eventDesc(status enum.CacheStatus, message string) string {
return fmt.Sprintf("%s - %s", status, message)
}
[02/20] incubator-trafficcontrol git commit: Fix TM2 threshold msg to
truncate precision
Posted by ne...@apache.org.
Fix TM2 threshold msg to truncate precision
Project: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/commit/ae137e5a
Tree: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/tree/ae137e5a
Diff: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/diff/ae137e5a
Branch: refs/heads/master
Commit: ae137e5aba5e7dd422ce1aeeac7577370fb282ac
Parents: f9115dc
Author: Robert Butts <ro...@gmail.com>
Authored: Tue Jan 24 14:23:10 2017 -0700
Committer: Dave Neuman <ne...@apache.org>
Committed: Wed Jan 25 10:29:46 2017 -0700
----------------------------------------------------------------------
.../experimental/traffic_monitor/health/cache.go | 10 +++++-----
1 file changed, 5 insertions(+), 5 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/ae137e5a/traffic_monitor/experimental/traffic_monitor/health/cache.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/health/cache.go b/traffic_monitor/experimental/traffic_monitor/health/cache.go
index fd3fcec..323d8e0 100644
--- a/traffic_monitor/experimental/traffic_monitor/health/cache.go
+++ b/traffic_monitor/experimental/traffic_monitor/health/cache.go
@@ -214,15 +214,15 @@ func setErr(newResult *cache.Result, err error) {
func exceedsThresholdMsg(stat string, threshold to.HealthThreshold, val float64) string {
switch threshold.Comparator {
case "=":
- return fmt.Sprintf("%s not equal (%f != %f)", stat, val, threshold.Val)
+ return fmt.Sprintf("%s not equal (%.2f != %.2f)", stat, val, threshold.Val)
case ">":
- return fmt.Sprintf("%s too low (%f < %f)", stat, val, threshold.Val)
+ return fmt.Sprintf("%s too low (%.2f < %.2f)", stat, val, threshold.Val)
case "<":
- return fmt.Sprintf("%s too high (%f > %f)", stat, val, threshold.Val)
+ return fmt.Sprintf("%s too high (%.2f > %.2f)", stat, val, threshold.Val)
case ">=":
- return fmt.Sprintf("%s too low (%f <= %f)", stat, val, threshold.Val)
+ return fmt.Sprintf("%s too low (%.2f <= %.2f)", stat, val, threshold.Val)
case "<=":
- return fmt.Sprintf("%s too high (%f >= %f)", stat, val, threshold.Val)
+ return fmt.Sprintf("%s too high (%.2f >= %.2f)", stat, val, threshold.Val)
default:
return fmt.Sprintf("ERROR: Invalid Threshold: %+v", threshold)
}
[08/20] incubator-trafficcontrol git commit: Add TM2 availability
poller to /api/cache-statuses
Posted by ne...@apache.org.
Add TM2 availability poller to /api/cache-statuses
Project: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/commit/34863edd
Tree: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/tree/34863edd
Diff: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/diff/34863edd
Branch: refs/heads/master
Commit: 34863edd4108b96dae4d1fd26d16d19fd033ca1b
Parents: 40d5bcb
Author: Robert Butts <ro...@gmail.com>
Authored: Fri Jan 20 16:00:54 2017 -0700
Committer: Dave Neuman <ne...@apache.org>
Committed: Wed Jan 25 10:29:46 2017 -0700
----------------------------------------------------------------------
.../experimental/traffic_monitor/cache/cache.go | 12 ++++++++++++
.../experimental/traffic_monitor/cache/data.go | 2 ++
.../traffic_monitor/manager/datarequest.go | 10 +++++++---
.../traffic_monitor/manager/healthresult.go | 13 -------------
.../traffic_monitor/manager/stathistory.go | 8 ++++----
5 files changed, 25 insertions(+), 20 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/34863edd/traffic_monitor/experimental/traffic_monitor/cache/cache.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/cache/cache.go b/traffic_monitor/experimental/traffic_monitor/cache/cache.go
index 541edaa..95aab84 100644
--- a/traffic_monitor/experimental/traffic_monitor/cache/cache.go
+++ b/traffic_monitor/experimental/traffic_monitor/cache/cache.go
@@ -87,6 +87,18 @@ type Result struct {
Available bool
}
+// HasStat returns whether the given stat is in the Result.
+func (result *Result) HasStat(stat string) bool {
+ computedStats := ComputedStats()
+ if _, ok := computedStats[stat]; ok {
+ return true // health poll has all computed stats
+ }
+ if _, ok := result.Astats.Ats[stat]; ok {
+ return true
+ }
+ return false
+}
+
// Vitals is the vitals data returned from a cache.
type Vitals struct {
LoadAvg float64
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/34863edd/traffic_monitor/experimental/traffic_monitor/cache/data.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/cache/data.go b/traffic_monitor/experimental/traffic_monitor/cache/data.go
index ee21907..bad75a0 100644
--- a/traffic_monitor/experimental/traffic_monitor/cache/data.go
+++ b/traffic_monitor/experimental/traffic_monitor/cache/data.go
@@ -38,6 +38,8 @@ type AvailableStatus struct {
Why string
// UnavailableStat is the stat whose threshold made the cache unavailable. If this is the empty string, the cache is unavailable for a non-threshold reason. This exists so a poller (health, stat) won't mark an unavailable cache as available if the stat whose threshold was reached isn't available on that poller.
UnavailableStat string
+ // Poller is the name of the poller which set this available status
+ Poller string
}
// CacheAvailableStatuses is the available status of each cache.
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/34863edd/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go b/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
index 2f42d74..2dffb41 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
@@ -63,9 +63,10 @@ type APIPeerStates struct {
// CacheStatus contains summary stat data about the given cache.
// TODO make fields nullable, so error fields can be omitted, letting API callers still get updates for unerrored fields
type CacheStatus struct {
- Type *string `json:"type,omitempty"`
- Status *string `json:"status,omitempty"`
- LoadAverage *float64 `json:"load_average,omitempty"`
+ Type *string `json:"type,omitempty"`
+ Status *string `json:"status,omitempty"`
+ StatusPoller *string `json:"status_poller,omitempty"`
+ LoadAverage *float64 `json:"load_average,omitempty"`
// QueryTimeMilliseconds is the time it took this app to perform a complete query and process the data, end-to-end, for the latest health query.
QueryTimeMilliseconds *int64 `json:"query_time_ms,omitempty"`
// HealthTimeMilliseconds is the time it took to make the HTTP request and get back the full response, for the latest health query.
@@ -971,6 +972,7 @@ func createCacheStatuses(
}
var status *string
+ var statusPoller *string
statusVal, ok := localCacheStatus[cacheName]
if !ok {
log.Warnf("cache not in statuses %s\n", cacheName)
@@ -987,6 +989,7 @@ func createCacheStatuses(
}
status = &statusString
+ statusPoller = &statusVal.Poller
}
cacheTypeStr := string(cacheType)
@@ -1002,6 +1005,7 @@ func createCacheStatuses(
BandwidthCapacityKbps: maxKbps,
ConnectionCount: connections,
Status: status,
+ StatusPoller: statusPoller,
}
}
return statii
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/34863edd/traffic_monitor/experimental/traffic_monitor/manager/healthresult.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/healthresult.go b/traffic_monitor/experimental/traffic_monitor/manager/healthresult.go
index 110f083..63248e8 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/healthresult.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/healthresult.go
@@ -245,19 +245,6 @@ func processHealthResult(
lastHealthDurationsThreadsafe.Set(lastHealthDurations)
}
-// resultHasStat returns whether the given stat is in the Result.
-// TODO move to cache?
-func resultHasStat(stat string, result cache.Result) bool {
- computedStats := cache.ComputedStats()
- if _, ok := computedStats[stat]; ok {
- return true // health poll has all computed stats
- }
- if _, ok := result.Astats.Ats[stat]; ok {
- return true
- }
- return false
-}
-
// calculateDeliveryServiceState calculates the state of delivery services from the new cache state data `cacheState` and the CRConfig data `deliveryServiceServers` and puts the calculated state in the outparam `deliveryServiceStates`
func CalculateDeliveryServiceState(deliveryServiceServers map[enum.DeliveryServiceName][]enum.CacheName, states peer.CRStatesThreadsafe) {
deliveryServices := states.GetDeliveryServices()
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/34863edd/traffic_monitor/experimental/traffic_monitor/manager/stathistory.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/stathistory.go b/traffic_monitor/experimental/traffic_monitor/manager/stathistory.go
index b44f357..c63584e 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/stathistory.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/stathistory.go
@@ -238,11 +238,10 @@ func calcAvailability(results []cache.Result, pollerName string, statResultHisto
}
isAvailable, whyAvailable, unavailableStat := health.EvalCache(cache.ToInfo(result), statResults, &mc)
- whyAvailable += " (" + pollerName + ")" // TODO move to field in AvailableStatus
// if the cache is now Available, and was previously unavailable due to a threshold, make sure this poller contains the stat which exceeded the threshold.
if previousStatus, hasPreviousStatus := localCacheStatuses[result.ID]; isAvailable && hasPreviousStatus && !previousStatus.Available && previousStatus.UnavailableStat != "" {
- if !resultHasStat(previousStatus.UnavailableStat, result) {
+ if !result.HasStat(previousStatus.UnavailableStat) {
return
}
}
@@ -251,11 +250,12 @@ func calcAvailability(results []cache.Result, pollerName string, statResultHisto
Status: mc.TrafficServer[string(result.ID)].Status,
Why: whyAvailable,
UnavailableStat: unavailableStat,
+ Poller: pollerName,
} // TODO move within localStates?
if available, ok := localStates.GetCache(result.ID); !ok || available.IsAvailable != isAvailable {
- log.Infof("Changing state for %s was: %t now: %t because %s error: %v", result.ID, available.IsAvailable, isAvailable, whyAvailable, result.Error)
- events.Add(cache.Event{Time: time.Now().Unix(), Description: whyAvailable, Name: result.ID, Hostname: result.ID, Type: toData.ServerTypes[result.ID].String(), Available: isAvailable})
+ log.Infof("Changing state for %s was: %t now: %t because %s poller: %v error: %v", result.ID, available.IsAvailable, isAvailable, whyAvailable, pollerName, result.Error)
+ events.Add(cache.Event{Time: time.Now().Unix(), Description: whyAvailable + " (" + pollerName + ")", Name: result.ID, Hostname: result.ID, Type: toData.ServerTypes[result.ID].String(), Available: isAvailable})
}
localStates.SetCache(result.ID, peer.IsAvailable{IsAvailable: isAvailable})
[11/20] incubator-trafficcontrol git commit: Fix TM2 unnecessary
conversions (lint)
Posted by ne...@apache.org.
Fix TM2 unnecessary conversions (lint)
Project: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/commit/1ffb3019
Tree: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/tree/1ffb3019
Diff: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/diff/1ffb3019
Branch: refs/heads/master
Commit: 1ffb301902b2ea2f8da243ed886e8da21ef240e5
Parents: 6760764
Author: Robert Butts <ro...@gmail.com>
Authored: Mon Jan 23 15:15:39 2017 -0700
Committer: Dave Neuman <ne...@apache.org>
Committed: Wed Jan 25 10:29:46 2017 -0700
----------------------------------------------------------------------
traffic_monitor/experimental/traffic_monitor/enum/enum.go | 2 +-
.../experimental/traffic_monitor/manager/datarequest.go | 4 ++--
2 files changed, 3 insertions(+), 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/1ffb3019/traffic_monitor/experimental/traffic_monitor/enum/enum.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/enum/enum.go b/traffic_monitor/experimental/traffic_monitor/enum/enum.go
index 44184df..0e43016 100644
--- a/traffic_monitor/experimental/traffic_monitor/enum/enum.go
+++ b/traffic_monitor/experimental/traffic_monitor/enum/enum.go
@@ -201,7 +201,7 @@ func ToNumeric(v interface{}) (float64, bool) {
case float32:
return float64(i), true
case float64:
- return float64(i), true
+ return i, true
case int:
return float64(i), true
case uint:
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/1ffb3019/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go b/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
index 47853b7..3cc4c4a 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
@@ -1185,8 +1185,8 @@ func createStatSummary(statResultHistory cache.ResultStatHistory, filter cache.F
}
ssStat := StatSummaryStat{}
msPerNs := int64(1000000)
- ssStat.StartTime = time.Time(statHistory[len(statHistory)-1].Time).UnixNano() / msPerNs
- ssStat.EndTime = time.Time(statHistory[0].Time).UnixNano() / msPerNs
+ ssStat.StartTime = statHistory[len(statHistory)-1].Time.UnixNano() / msPerNs
+ ssStat.EndTime = statHistory[0].Time.UnixNano() / msPerNs
oldestVal, isOldestValNumeric := enum.ToNumeric(statHistory[len(statHistory)-1].Val)
newestVal, isNewestValNumeric := enum.ToNumeric(statHistory[0].Val)
if !isOldestValNumeric || !isNewestValNumeric {
[04/20] incubator-trafficcontrol git commit: Change TM2 cache handler
result chan to be private
Posted by ne...@apache.org.
Change TM2 cache handler result chan to be private
Project: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/commit/f31ffc73
Tree: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/tree/f31ffc73
Diff: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/diff/f31ffc73
Branch: refs/heads/master
Commit: f31ffc733a0b45f26ea09f8e58686bcee55fc65d
Parents: b2cb736
Author: Robert Butts <ro...@gmail.com>
Authored: Mon Jan 23 10:40:35 2017 -0700
Committer: Dave Neuman <ne...@apache.org>
Committed: Wed Jan 25 10:29:46 2017 -0700
----------------------------------------------------------------------
.../experimental/traffic_monitor/cache/cache.go | 26 +++++++++++---------
.../traffic_monitor/manager/manager.go | 4 +--
2 files changed, 16 insertions(+), 14 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/f31ffc73/traffic_monitor/experimental/traffic_monitor/cache/cache.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/cache/cache.go b/traffic_monitor/experimental/traffic_monitor/cache/cache.go
index 95aab84..a850102 100644
--- a/traffic_monitor/experimental/traffic_monitor/cache/cache.go
+++ b/traffic_monitor/experimental/traffic_monitor/cache/cache.go
@@ -40,25 +40,28 @@ import (
// Handler is a cache handler, which fulfills the common/handler `Handler` interface.
type Handler struct {
- ResultChannel chan Result
+ resultChan chan Result
Notify int
ToData *todata.TODataThreadsafe
PeerStates *peer.CRStatesPeersThreadsafe
MultipleSpaceRegex *regexp.Regexp
}
-// NewHandler returns a new cache handler. Note this handler does NOT precomputes stat data before calling ResultChannel, and Result.Precomputed will be nil
-// TODO change this to take the ResultChan. It doesn't make sense for the Handler to 'own' the Result Chan.
+func (h Handler) ResultChan() <-chan Result {
+ return h.resultChan
+}
+
+// NewHandler returns a new cache handler. Note this handler does NOT precomputes stat data before calling ResultChan, and Result.Precomputed will be nil
func NewHandler() Handler {
- return Handler{ResultChannel: make(chan Result), MultipleSpaceRegex: regexp.MustCompile(" +")}
+ return Handler{resultChan: make(chan Result), MultipleSpaceRegex: regexp.MustCompile(" +")}
}
-// NewPrecomputeHandler constructs a new cache Handler, which precomputes stat data and populates result.Precomputed before passing to ResultChannel.
+// NewPrecomputeHandler constructs a new cache Handler, which precomputes stat data and populates result.Precomputed before passing to ResultChan.
func NewPrecomputeHandler(toData todata.TODataThreadsafe, peerStates peer.CRStatesPeersThreadsafe) Handler {
- return Handler{ResultChannel: make(chan Result), MultipleSpaceRegex: regexp.MustCompile(" +"), ToData: &toData, PeerStates: &peerStates}
+ return Handler{resultChan: make(chan Result), MultipleSpaceRegex: regexp.MustCompile(" +"), ToData: &toData, PeerStates: &peerStates}
}
-// Precompute returns whether this handler precomputes data before passing the result to the ResultChannel
+// Precompute returns whether this handler precomputes data before passing the result to the ResultChan
func (handler Handler) Precompute() bool {
return handler.ToData != nil && handler.PeerStates != nil
}
@@ -311,14 +314,14 @@ func (handler Handler) Handle(id string, r io.Reader, reqTime time.Duration, req
if reqErr != nil {
log.Errorf("%v handler given error '%v'\n", id, reqErr) // error here, in case the thing that called Handle didn't error
result.Error = reqErr
- handler.ResultChannel <- result
+ handler.resultChan <- result
return
}
if r == nil {
log.Errorf("%v handle reader nil\n", id)
result.Error = fmt.Errorf("handler got nil reader")
- handler.ResultChannel <- result
+ handler.resultChan <- result
return
}
@@ -328,7 +331,7 @@ func (handler Handler) Handle(id string, r io.Reader, reqTime time.Duration, req
if decodeErr := json.NewDecoder(r).Decode(&result.Astats); decodeErr != nil {
log.Errorf("%s procnetdev decode error '%v'\n", id, decodeErr)
result.Error = decodeErr
- handler.ResultChannel <- result
+ handler.resultChan <- result
return
}
@@ -355,7 +358,7 @@ func (handler Handler) Handle(id string, r io.Reader, reqTime time.Duration, req
log.Debugf("poll %v %v handle precompute end\n", pollID, time.Now())
}
log.Debugf("poll %v %v handle write start\n", pollID, time.Now())
- handler.ResultChannel <- result
+ handler.resultChan <- result
log.Debugf("poll %v %v handle end\n", pollID, time.Now())
}
@@ -412,7 +415,6 @@ func (handler Handler) precompute(result Result) Result {
}
// processStat and its subsidiary functions act as a State Machine, flowing the stat thru states for each "." component of the stat name
-// TODO fix this being crazy slow. THIS IS THE BOTTLENECK
func processStat(server enum.CacheName, stats map[enum.DeliveryServiceName]dsdata.Stat, toData todata.TOData, stat string, value interface{}, timeReceived time.Time) (map[enum.DeliveryServiceName]dsdata.Stat, error) {
parts := strings.Split(stat, ".")
if len(parts) < 1 {
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/f31ffc73/traffic_monitor/experimental/traffic_monitor/manager/manager.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/manager.go b/traffic_monitor/experimental/traffic_monitor/manager/manager.go
index dba10ad..b5e3d49 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/manager.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/manager.go
@@ -113,7 +113,7 @@ func Start(opsConfigFile string, cfg config.Config, staticAppData StaticAppData)
)
statInfoHistory, statResultHistory, statMaxKbpses, _, lastKbpsStats, dsStats, unpolledCaches, localCacheStatus := StartStatHistoryManager(
- cacheStatHandler.ResultChannel,
+ cacheStatHandler.ResultChan(),
localStates,
combinedStates,
toData,
@@ -125,7 +125,7 @@ func Start(opsConfigFile string, cfg config.Config, staticAppData StaticAppData)
)
lastHealthDurations, healthHistory := StartHealthResultManager(
- cacheHealthHandler.ResultChannel,
+ cacheHealthHandler.ResultChan(),
toData,
localStates,
monitorConfig,
[03/20] incubator-trafficcontrol git commit: Rename TM2
manager/stathistory.go to stat.go
Posted by ne...@apache.org.
Rename TM2 manager/stathistory.go to stat.go
Project: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/commit/0d898cc6
Tree: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/tree/0d898cc6
Diff: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/diff/0d898cc6
Branch: refs/heads/master
Commit: 0d898cc6d8d2da80626eb5e5222a14026fa76ea2
Parents: b575ec1
Author: Robert Butts <ro...@gmail.com>
Authored: Tue Jan 24 10:45:40 2017 -0700
Committer: Dave Neuman <ne...@apache.org>
Committed: Wed Jan 25 10:29:46 2017 -0700
----------------------------------------------------------------------
.../traffic_monitor/manager/stat.go | 265 +++++++++++++++++++
.../traffic_monitor/manager/stathistory.go | 265 -------------------
2 files changed, 265 insertions(+), 265 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/0d898cc6/traffic_monitor/experimental/traffic_monitor/manager/stat.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/stat.go b/traffic_monitor/experimental/traffic_monitor/manager/stat.go
new file mode 100644
index 0000000..a4a9b81
--- /dev/null
+++ b/traffic_monitor/experimental/traffic_monitor/manager/stat.go
@@ -0,0 +1,265 @@
+package manager
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import (
+ "time"
+
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/common/log"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/cache"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/config"
+ ds "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/deliveryservice"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/enum"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/health"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/peer"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/threadsafe"
+ todata "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/trafficopsdata"
+ to "github.com/apache/incubator-trafficcontrol/traffic_ops/client"
+)
+
+func pruneHistory(history []cache.Result, limit uint64) []cache.Result {
+ if uint64(len(history)) > limit {
+ history = history[:limit-1]
+ }
+ return history
+}
+
+func getNewCaches(localStates peer.CRStatesThreadsafe, monitorConfigTS TrafficMonitorConfigMapThreadsafe) map[enum.CacheName]struct{} {
+ monitorConfig := monitorConfigTS.Get()
+ caches := map[enum.CacheName]struct{}{}
+ for cacheName := range localStates.GetCaches() {
+ // ONLINE and OFFLINE caches are not polled.
+ // TODO add a function IsPolled() which can be called by this and the monitorConfig func which sets the polling, to prevent updating in one place breaking the other.
+ if ts, ok := monitorConfig.TrafficServer[string(cacheName)]; !ok || ts.Status == "ONLINE" || ts.Status == "OFFLINE" {
+ continue
+ }
+ caches[cacheName] = struct{}{}
+ }
+ return caches
+}
+
+// StartStatHistoryManager fetches the full statistics data from ATS Astats. This includes everything needed for all calculations, such as Delivery Services. This is expensive, though, and may be hard on ATS, so it should poll less often.
+// For a fast 'is it alive' poll, use the Health Result Manager poll.
+// Returns the stat history, the duration between the stat poll for each cache, the last Kbps data, the calculated Delivery Service stats, and the unpolled caches list.
+func StartStatHistoryManager(
+ cacheStatChan <-chan cache.Result,
+ localStates peer.CRStatesThreadsafe,
+ combinedStates peer.CRStatesThreadsafe,
+ toData todata.TODataThreadsafe,
+ cachesChanged <-chan struct{},
+ errorCount threadsafe.Uint,
+ cfg config.Config,
+ monitorConfig TrafficMonitorConfigMapThreadsafe,
+ events threadsafe.Events,
+) (threadsafe.ResultInfoHistory, threadsafe.ResultStatHistory, threadsafe.CacheKbpses, DurationMapThreadsafe, threadsafe.LastStats, threadsafe.DSStatsReader, threadsafe.UnpolledCaches, threadsafe.CacheAvailableStatus) {
+ statInfoHistory := threadsafe.NewResultInfoHistory()
+ statResultHistory := threadsafe.NewResultStatHistory()
+ statMaxKbpses := threadsafe.NewCacheKbpses()
+ lastStatDurations := NewDurationMapThreadsafe()
+ lastStatEndTimes := map[enum.CacheName]time.Time{}
+ lastStats := threadsafe.NewLastStats()
+ dsStats := threadsafe.NewDSStats()
+ unpolledCaches := threadsafe.NewUnpolledCaches()
+ tickInterval := cfg.StatFlushInterval
+ localCacheStatus := threadsafe.NewCacheAvailableStatus()
+
+ precomputedData := map[enum.CacheName]cache.PrecomputedData{}
+ lastResults := map[enum.CacheName]cache.Result{}
+
+ process := func(results []cache.Result) {
+ processStatResults(results, statInfoHistory, statResultHistory, statMaxKbpses, combinedStates.Get(), lastStats, toData.Get(), errorCount, dsStats, lastStatEndTimes, lastStatDurations, unpolledCaches, monitorConfig.Get(), precomputedData, lastResults, localStates, events, localCacheStatus)
+ }
+
+ go func() {
+ var ticker *time.Ticker
+ <-cachesChanged // wait for the signal that localStates have been set
+ unpolledCaches.SetNewCaches(getNewCaches(localStates, monitorConfig))
+
+ for {
+ var results []cache.Result
+ results = append(results, <-cacheStatChan)
+ if ticker != nil {
+ ticker.Stop()
+ }
+ ticker = time.NewTicker(tickInterval)
+ innerLoop:
+ for {
+ select {
+ case <-cachesChanged:
+ unpolledCaches.SetNewCaches(getNewCaches(localStates, monitorConfig))
+ case <-ticker.C:
+ log.Warnf("StatHistoryManager flushing queued results\n")
+ process(results)
+ break innerLoop
+ default:
+ select {
+ case r := <-cacheStatChan:
+ results = append(results, r)
+ default:
+ process(results)
+ break innerLoop
+ }
+ }
+ }
+ }
+ }()
+ return statInfoHistory, statResultHistory, statMaxKbpses, lastStatDurations, lastStats, &dsStats, unpolledCaches, localCacheStatus
+}
+
+// processStatResults processes the given results, creating and setting DSStats, LastStats, and other stats. Note this is NOT threadsafe, and MUST NOT be called from multiple threads.
+func processStatResults(
+ results []cache.Result,
+ statInfoHistoryThreadsafe threadsafe.ResultInfoHistory,
+ statResultHistoryThreadsafe threadsafe.ResultStatHistory,
+ statMaxKbpsesThreadsafe threadsafe.CacheKbpses,
+ combinedStates peer.Crstates,
+ lastStats threadsafe.LastStats,
+ toData todata.TOData,
+ errorCount threadsafe.Uint,
+ dsStats threadsafe.DSStats,
+ lastStatEndTimes map[enum.CacheName]time.Time,
+ lastStatDurationsThreadsafe DurationMapThreadsafe,
+ unpolledCaches threadsafe.UnpolledCaches,
+ mc to.TrafficMonitorConfigMap,
+ precomputedData map[enum.CacheName]cache.PrecomputedData,
+ lastResults map[enum.CacheName]cache.Result,
+ localStates peer.CRStatesThreadsafe,
+ events threadsafe.Events,
+ localCacheStatusThreadsafe threadsafe.CacheAvailableStatus,
+) {
+ if len(results) == 0 {
+ return
+ }
+ defer func() {
+ for _, r := range results {
+ // log.Debugf("poll %v %v statfinish\n", result.PollID, endTime)
+ r.PollFinished <- r.PollID
+ }
+ }()
+
+ // setting the statHistory could be put in a goroutine concurrent with `ds.CreateStats`, if it were slow
+ statInfoHistory := statInfoHistoryThreadsafe.Get().Copy()
+ statResultHistory := statResultHistoryThreadsafe.Get().Copy()
+ statMaxKbpses := statMaxKbpsesThreadsafe.Get().Copy()
+
+ for i, result := range results {
+ maxStats := uint64(mc.Profile[mc.TrafficServer[string(result.ID)].Profile].Parameters.HistoryCount)
+ if maxStats < 1 {
+ log.Warnf("processStatResults got history count %v for %v, setting to 1\n", maxStats, result.ID)
+ maxStats = 1
+ }
+
+ // TODO determine if we want to add results with errors, or just print the errors now and don't add them.
+ if lastResult, ok := lastResults[result.ID]; ok && result.Error == nil {
+ health.GetVitals(&result, &lastResult, &mc) // TODO precompute
+ if result.Error == nil {
+ results[i] = result
+ } else {
+ log.Errorf("stat poll getting vitals for %v: %v\n", result.ID, result.Error)
+ }
+ }
+ statInfoHistory.Add(result, maxStats)
+ statResultHistory.Add(result, maxStats)
+ // Don't add errored maxes or precomputed DSStats
+ if result.Error == nil {
+ // max and precomputed always contain the latest result from each cache
+ statMaxKbpses.AddMax(result)
+ // if we failed to compute the OutBytes, keep the outbytes of the last result.
+ if result.PrecomputedData.OutBytes == 0 {
+ result.PrecomputedData.OutBytes = precomputedData[result.ID].OutBytes
+ }
+ precomputedData[result.ID] = result.PrecomputedData
+
+ }
+ lastResults[result.ID] = result
+ }
+ statInfoHistoryThreadsafe.Set(statInfoHistory)
+ statResultHistoryThreadsafe.Set(statResultHistory)
+ statMaxKbpsesThreadsafe.Set(statMaxKbpses)
+
+ for _, result := range results {
+ log.Debugf("poll %v %v CreateStats start\n", result.PollID, time.Now())
+ }
+
+ newDsStats, newLastStats, err := ds.CreateStats(precomputedData, toData, combinedStates, lastStats.Get().Copy(), time.Now(), mc)
+
+ for _, result := range results {
+ log.Debugf("poll %v %v CreateStats end\n", result.PollID, time.Now())
+ }
+
+ if err != nil {
+ errorCount.Inc()
+ log.Errorf("getting deliveryservice: %v\n", err)
+ } else {
+ dsStats.Set(newDsStats)
+ lastStats.Set(newLastStats)
+ }
+
+ calcAvailability(results, "stat", statResultHistory, mc, toData, localCacheStatusThreadsafe, localStates, events)
+
+ endTime := time.Now()
+ lastStatDurations := lastStatDurationsThreadsafe.Get().Copy()
+ for _, result := range results {
+ if lastStatStart, ok := lastStatEndTimes[result.ID]; ok {
+ d := time.Since(lastStatStart)
+ lastStatDurations[result.ID] = d
+ }
+ lastStatEndTimes[result.ID] = endTime
+ }
+ lastStatDurationsThreadsafe.Set(lastStatDurations)
+ unpolledCaches.SetPolled(results, lastStats.Get())
+}
+
+// calcAvailability calculates the availability of the cache, from the given result. Availability is stored in `localCacheStatus` and `localStates`, and if the status changed an event is added to `events`. statResultHistory may be nil, for pollers which don't poll stats.
+// TODO add enum for poller names?
+func calcAvailability(results []cache.Result, pollerName string, statResultHistory cache.ResultStatHistory, mc to.TrafficMonitorConfigMap, toData todata.TOData, localCacheStatusThreadsafe threadsafe.CacheAvailableStatus, localStates peer.CRStatesThreadsafe, events threadsafe.Events) {
+ localCacheStatuses := localCacheStatusThreadsafe.Get().Copy()
+ for _, result := range results {
+ statResults := cache.ResultStatValHistory(nil)
+ if statResultHistory != nil {
+ statResults = statResultHistory[result.ID]
+ }
+
+ isAvailable, whyAvailable, unavailableStat := health.EvalCache(cache.ToInfo(result), statResults, &mc)
+
+ // if the cache is now Available, and was previously unavailable due to a threshold, make sure this poller contains the stat which exceeded the threshold.
+ if previousStatus, hasPreviousStatus := localCacheStatuses[result.ID]; isAvailable && hasPreviousStatus && !previousStatus.Available && previousStatus.UnavailableStat != "" {
+ if !result.HasStat(previousStatus.UnavailableStat) {
+ return
+ }
+ }
+ localCacheStatuses[result.ID] = cache.AvailableStatus{
+ Available: isAvailable,
+ Status: mc.TrafficServer[string(result.ID)].Status,
+ Why: whyAvailable,
+ UnavailableStat: unavailableStat,
+ Poller: pollerName,
+ } // TODO move within localStates?
+
+ if available, ok := localStates.GetCache(result.ID); !ok || available.IsAvailable != isAvailable {
+ log.Infof("Changing state for %s was: %t now: %t because %s poller: %v error: %v", result.ID, available.IsAvailable, isAvailable, whyAvailable, pollerName, result.Error)
+ events.Add(health.Event{Time: time.Now(), Description: whyAvailable + " (" + pollerName + ")", Name: string(result.ID), Hostname: string(result.ID), Type: toData.ServerTypes[result.ID].String(), Available: isAvailable})
+ }
+
+ localStates.SetCache(result.ID, peer.IsAvailable{IsAvailable: isAvailable})
+ }
+ CalculateDeliveryServiceState(toData.DeliveryServiceServers, localStates)
+ localCacheStatusThreadsafe.Set(localCacheStatuses)
+}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/0d898cc6/traffic_monitor/experimental/traffic_monitor/manager/stathistory.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/stathistory.go b/traffic_monitor/experimental/traffic_monitor/manager/stathistory.go
deleted file mode 100644
index a4a9b81..0000000
--- a/traffic_monitor/experimental/traffic_monitor/manager/stathistory.go
+++ /dev/null
@@ -1,265 +0,0 @@
-package manager
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-import (
- "time"
-
- "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/common/log"
- "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/cache"
- "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/config"
- ds "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/deliveryservice"
- "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/enum"
- "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/health"
- "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/peer"
- "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/threadsafe"
- todata "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/trafficopsdata"
- to "github.com/apache/incubator-trafficcontrol/traffic_ops/client"
-)
-
-func pruneHistory(history []cache.Result, limit uint64) []cache.Result {
- if uint64(len(history)) > limit {
- history = history[:limit-1]
- }
- return history
-}
-
-func getNewCaches(localStates peer.CRStatesThreadsafe, monitorConfigTS TrafficMonitorConfigMapThreadsafe) map[enum.CacheName]struct{} {
- monitorConfig := monitorConfigTS.Get()
- caches := map[enum.CacheName]struct{}{}
- for cacheName := range localStates.GetCaches() {
- // ONLINE and OFFLINE caches are not polled.
- // TODO add a function IsPolled() which can be called by this and the monitorConfig func which sets the polling, to prevent updating in one place breaking the other.
- if ts, ok := monitorConfig.TrafficServer[string(cacheName)]; !ok || ts.Status == "ONLINE" || ts.Status == "OFFLINE" {
- continue
- }
- caches[cacheName] = struct{}{}
- }
- return caches
-}
-
-// StartStatHistoryManager fetches the full statistics data from ATS Astats. This includes everything needed for all calculations, such as Delivery Services. This is expensive, though, and may be hard on ATS, so it should poll less often.
-// For a fast 'is it alive' poll, use the Health Result Manager poll.
-// Returns the stat history, the duration between the stat poll for each cache, the last Kbps data, the calculated Delivery Service stats, and the unpolled caches list.
-func StartStatHistoryManager(
- cacheStatChan <-chan cache.Result,
- localStates peer.CRStatesThreadsafe,
- combinedStates peer.CRStatesThreadsafe,
- toData todata.TODataThreadsafe,
- cachesChanged <-chan struct{},
- errorCount threadsafe.Uint,
- cfg config.Config,
- monitorConfig TrafficMonitorConfigMapThreadsafe,
- events threadsafe.Events,
-) (threadsafe.ResultInfoHistory, threadsafe.ResultStatHistory, threadsafe.CacheKbpses, DurationMapThreadsafe, threadsafe.LastStats, threadsafe.DSStatsReader, threadsafe.UnpolledCaches, threadsafe.CacheAvailableStatus) {
- statInfoHistory := threadsafe.NewResultInfoHistory()
- statResultHistory := threadsafe.NewResultStatHistory()
- statMaxKbpses := threadsafe.NewCacheKbpses()
- lastStatDurations := NewDurationMapThreadsafe()
- lastStatEndTimes := map[enum.CacheName]time.Time{}
- lastStats := threadsafe.NewLastStats()
- dsStats := threadsafe.NewDSStats()
- unpolledCaches := threadsafe.NewUnpolledCaches()
- tickInterval := cfg.StatFlushInterval
- localCacheStatus := threadsafe.NewCacheAvailableStatus()
-
- precomputedData := map[enum.CacheName]cache.PrecomputedData{}
- lastResults := map[enum.CacheName]cache.Result{}
-
- process := func(results []cache.Result) {
- processStatResults(results, statInfoHistory, statResultHistory, statMaxKbpses, combinedStates.Get(), lastStats, toData.Get(), errorCount, dsStats, lastStatEndTimes, lastStatDurations, unpolledCaches, monitorConfig.Get(), precomputedData, lastResults, localStates, events, localCacheStatus)
- }
-
- go func() {
- var ticker *time.Ticker
- <-cachesChanged // wait for the signal that localStates have been set
- unpolledCaches.SetNewCaches(getNewCaches(localStates, monitorConfig))
-
- for {
- var results []cache.Result
- results = append(results, <-cacheStatChan)
- if ticker != nil {
- ticker.Stop()
- }
- ticker = time.NewTicker(tickInterval)
- innerLoop:
- for {
- select {
- case <-cachesChanged:
- unpolledCaches.SetNewCaches(getNewCaches(localStates, monitorConfig))
- case <-ticker.C:
- log.Warnf("StatHistoryManager flushing queued results\n")
- process(results)
- break innerLoop
- default:
- select {
- case r := <-cacheStatChan:
- results = append(results, r)
- default:
- process(results)
- break innerLoop
- }
- }
- }
- }
- }()
- return statInfoHistory, statResultHistory, statMaxKbpses, lastStatDurations, lastStats, &dsStats, unpolledCaches, localCacheStatus
-}
-
-// processStatResults processes the given results, creating and setting DSStats, LastStats, and other stats. Note this is NOT threadsafe, and MUST NOT be called from multiple threads.
-func processStatResults(
- results []cache.Result,
- statInfoHistoryThreadsafe threadsafe.ResultInfoHistory,
- statResultHistoryThreadsafe threadsafe.ResultStatHistory,
- statMaxKbpsesThreadsafe threadsafe.CacheKbpses,
- combinedStates peer.Crstates,
- lastStats threadsafe.LastStats,
- toData todata.TOData,
- errorCount threadsafe.Uint,
- dsStats threadsafe.DSStats,
- lastStatEndTimes map[enum.CacheName]time.Time,
- lastStatDurationsThreadsafe DurationMapThreadsafe,
- unpolledCaches threadsafe.UnpolledCaches,
- mc to.TrafficMonitorConfigMap,
- precomputedData map[enum.CacheName]cache.PrecomputedData,
- lastResults map[enum.CacheName]cache.Result,
- localStates peer.CRStatesThreadsafe,
- events threadsafe.Events,
- localCacheStatusThreadsafe threadsafe.CacheAvailableStatus,
-) {
- if len(results) == 0 {
- return
- }
- defer func() {
- for _, r := range results {
- // log.Debugf("poll %v %v statfinish\n", result.PollID, endTime)
- r.PollFinished <- r.PollID
- }
- }()
-
- // setting the statHistory could be put in a goroutine concurrent with `ds.CreateStats`, if it were slow
- statInfoHistory := statInfoHistoryThreadsafe.Get().Copy()
- statResultHistory := statResultHistoryThreadsafe.Get().Copy()
- statMaxKbpses := statMaxKbpsesThreadsafe.Get().Copy()
-
- for i, result := range results {
- maxStats := uint64(mc.Profile[mc.TrafficServer[string(result.ID)].Profile].Parameters.HistoryCount)
- if maxStats < 1 {
- log.Warnf("processStatResults got history count %v for %v, setting to 1\n", maxStats, result.ID)
- maxStats = 1
- }
-
- // TODO determine if we want to add results with errors, or just print the errors now and don't add them.
- if lastResult, ok := lastResults[result.ID]; ok && result.Error == nil {
- health.GetVitals(&result, &lastResult, &mc) // TODO precompute
- if result.Error == nil {
- results[i] = result
- } else {
- log.Errorf("stat poll getting vitals for %v: %v\n", result.ID, result.Error)
- }
- }
- statInfoHistory.Add(result, maxStats)
- statResultHistory.Add(result, maxStats)
- // Don't add errored maxes or precomputed DSStats
- if result.Error == nil {
- // max and precomputed always contain the latest result from each cache
- statMaxKbpses.AddMax(result)
- // if we failed to compute the OutBytes, keep the outbytes of the last result.
- if result.PrecomputedData.OutBytes == 0 {
- result.PrecomputedData.OutBytes = precomputedData[result.ID].OutBytes
- }
- precomputedData[result.ID] = result.PrecomputedData
-
- }
- lastResults[result.ID] = result
- }
- statInfoHistoryThreadsafe.Set(statInfoHistory)
- statResultHistoryThreadsafe.Set(statResultHistory)
- statMaxKbpsesThreadsafe.Set(statMaxKbpses)
-
- for _, result := range results {
- log.Debugf("poll %v %v CreateStats start\n", result.PollID, time.Now())
- }
-
- newDsStats, newLastStats, err := ds.CreateStats(precomputedData, toData, combinedStates, lastStats.Get().Copy(), time.Now(), mc)
-
- for _, result := range results {
- log.Debugf("poll %v %v CreateStats end\n", result.PollID, time.Now())
- }
-
- if err != nil {
- errorCount.Inc()
- log.Errorf("getting deliveryservice: %v\n", err)
- } else {
- dsStats.Set(newDsStats)
- lastStats.Set(newLastStats)
- }
-
- calcAvailability(results, "stat", statResultHistory, mc, toData, localCacheStatusThreadsafe, localStates, events)
-
- endTime := time.Now()
- lastStatDurations := lastStatDurationsThreadsafe.Get().Copy()
- for _, result := range results {
- if lastStatStart, ok := lastStatEndTimes[result.ID]; ok {
- d := time.Since(lastStatStart)
- lastStatDurations[result.ID] = d
- }
- lastStatEndTimes[result.ID] = endTime
- }
- lastStatDurationsThreadsafe.Set(lastStatDurations)
- unpolledCaches.SetPolled(results, lastStats.Get())
-}
-
-// calcAvailability calculates the availability of the cache, from the given result. Availability is stored in `localCacheStatus` and `localStates`, and if the status changed an event is added to `events`. statResultHistory may be nil, for pollers which don't poll stats.
-// TODO add enum for poller names?
-func calcAvailability(results []cache.Result, pollerName string, statResultHistory cache.ResultStatHistory, mc to.TrafficMonitorConfigMap, toData todata.TOData, localCacheStatusThreadsafe threadsafe.CacheAvailableStatus, localStates peer.CRStatesThreadsafe, events threadsafe.Events) {
- localCacheStatuses := localCacheStatusThreadsafe.Get().Copy()
- for _, result := range results {
- statResults := cache.ResultStatValHistory(nil)
- if statResultHistory != nil {
- statResults = statResultHistory[result.ID]
- }
-
- isAvailable, whyAvailable, unavailableStat := health.EvalCache(cache.ToInfo(result), statResults, &mc)
-
- // if the cache is now Available, and was previously unavailable due to a threshold, make sure this poller contains the stat which exceeded the threshold.
- if previousStatus, hasPreviousStatus := localCacheStatuses[result.ID]; isAvailable && hasPreviousStatus && !previousStatus.Available && previousStatus.UnavailableStat != "" {
- if !result.HasStat(previousStatus.UnavailableStat) {
- return
- }
- }
- localCacheStatuses[result.ID] = cache.AvailableStatus{
- Available: isAvailable,
- Status: mc.TrafficServer[string(result.ID)].Status,
- Why: whyAvailable,
- UnavailableStat: unavailableStat,
- Poller: pollerName,
- } // TODO move within localStates?
-
- if available, ok := localStates.GetCache(result.ID); !ok || available.IsAvailable != isAvailable {
- log.Infof("Changing state for %s was: %t now: %t because %s poller: %v error: %v", result.ID, available.IsAvailable, isAvailable, whyAvailable, pollerName, result.Error)
- events.Add(health.Event{Time: time.Now(), Description: whyAvailable + " (" + pollerName + ")", Name: string(result.ID), Hostname: string(result.ID), Type: toData.ServerTypes[result.ID].String(), Available: isAvailable})
- }
-
- localStates.SetCache(result.ID, peer.IsAvailable{IsAvailable: isAvailable})
- }
- CalculateDeliveryServiceState(toData.DeliveryServiceServers, localStates)
- localCacheStatusThreadsafe.Set(localCacheStatuses)
-}
[14/20] incubator-trafficcontrol git commit: Fix TM2 compile issue,
rename health/cache.go
Posted by ne...@apache.org.
Fix TM2 compile issue, rename health/cache.go
Project: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/commit/3db9dd7b
Tree: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/tree/3db9dd7b
Diff: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/diff/3db9dd7b
Branch: refs/heads/master
Commit: 3db9dd7bd952249c071bc0bec238cd6d89ef9e21
Parents: 4fec599
Author: Robert Butts <ro...@gmail.com>
Authored: Tue Jan 24 10:41:37 2017 -0700
Committer: Dave Neuman <ne...@apache.org>
Committed: Wed Jan 25 10:29:46 2017 -0700
----------------------------------------------------------------------
.../traffic_monitor/health/cache.go | 210 +++++++++++++++++++
.../traffic_monitor/health/cache_health.go | 210 -------------------
.../traffic_monitor/manager/stathistory.go | 2 +-
3 files changed, 211 insertions(+), 211 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/3db9dd7b/traffic_monitor/experimental/traffic_monitor/health/cache.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/health/cache.go b/traffic_monitor/experimental/traffic_monitor/health/cache.go
new file mode 100644
index 0000000..0a00838
--- /dev/null
+++ b/traffic_monitor/experimental/traffic_monitor/health/cache.go
@@ -0,0 +1,210 @@
+package health
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import (
+ "fmt"
+ "strconv"
+ "strings"
+
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/common/log"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/cache"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/enum"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/peer"
+ to "github.com/apache/incubator-trafficcontrol/traffic_ops/client"
+)
+
+func setError(newResult *cache.Result, err error) {
+ newResult.Error = err
+ newResult.Available = false
+}
+
+// GetVitals Gets the vitals to decide health on in the right format
+func GetVitals(newResult *cache.Result, prevResult *cache.Result, mc *to.TrafficMonitorConfigMap) {
+ if newResult.Error != nil {
+ log.Errorf("cache_health.GetVitals() called with an errored Result!")
+ return
+ }
+ // proc.loadavg -- we're using the 1 minute average (!?)
+ // value looks like: "0.20 0.07 0.07 1/967 29536" (without the quotes)
+ loadAverages := strings.Fields(newResult.Astats.System.ProcLoadavg)
+ if len(loadAverages) > 0 {
+ oneMinAvg, err := strconv.ParseFloat(loadAverages[0], 64)
+ if err != nil {
+ setError(newResult, fmt.Errorf("Error converting load average string '%s': %v", newResult.Astats.System.ProcLoadavg, err))
+ return
+ }
+ newResult.Vitals.LoadAvg = oneMinAvg
+ } else {
+ setError(newResult, fmt.Errorf("Can't make sense of '%s' as a load average for %s", newResult.Astats.System.ProcLoadavg, newResult.ID))
+ return
+ }
+
+ // proc.net.dev -- need to compare to prevSample
+ // value looks like
+ // "bond0:8495786321839 31960528603 0 0 0 0 0 2349716 143283576747316 101104535041 0 0 0 0 0 0"
+ // (without the quotes)
+ parts := strings.Split(newResult.Astats.System.ProcNetDev, ":")
+ if len(parts) > 1 {
+ numbers := strings.Fields(parts[1])
+ var err error
+ newResult.Vitals.BytesOut, err = strconv.ParseInt(numbers[8], 10, 64)
+ if err != nil {
+ setError(newResult, fmt.Errorf("Error converting BytesOut from procnetdev: %v", err))
+ return
+ }
+ newResult.Vitals.BytesIn, err = strconv.ParseInt(numbers[0], 10, 64)
+ if err != nil {
+ setError(newResult, fmt.Errorf("Error converting BytesIn from procnetdev: %v", err))
+ return
+ }
+ if prevResult != nil && prevResult.Vitals.BytesOut != 0 {
+ elapsedTimeInSecs := float64(newResult.Time.UnixNano()-prevResult.Time.UnixNano()) / 1000000000
+ newResult.Vitals.KbpsOut = int64(float64(((newResult.Vitals.BytesOut - prevResult.Vitals.BytesOut) * 8 / 1000)) / elapsedTimeInSecs)
+ } else {
+ // log.Infoln("prevResult == nil for id " + newResult.Id + ". Hope we're just starting up?")
+ }
+ } else {
+ setError(newResult, fmt.Errorf("Error parsing procnetdev: no fields found"))
+ return
+ }
+
+ // inf.speed -- value looks like "10000" (without the quotes) so it is in Mbps.
+ // TODO JvD: Should we really be running this code every second for every cache polled????? I don't think so.
+ interfaceBandwidth := newResult.Astats.System.InfSpeed
+ newResult.Vitals.MaxKbpsOut = int64(interfaceBandwidth) * 1000
+
+ // log.Infoln(newResult.Id, "BytesOut", newResult.Vitals.BytesOut, "BytesIn", newResult.Vitals.BytesIn, "Kbps", newResult.Vitals.KbpsOut, "max", newResult.Vitals.MaxKbpsOut)
+}
+
+// EvalCache returns whether the given cache should be marked available, a string describing why, and which stat exceeded a threshold. The `stats` may be nil, for pollers which don't poll stats.
+// The availability of EvalCache MAY NOT be used to directly set the cache's local availability, because the threshold stats may not be part of the poller which produced the result. Rather, if the cache was previously unavailable from a threshold, it must be verified that threshold stat is in the results before setting the cache to available.
+// TODO change to return a `cache.AvailableStatus`
+func EvalCache(result cache.ResultInfo, resultStats cache.ResultStatValHistory, mc *to.TrafficMonitorConfigMap) (bool, string, string) {
+ serverInfo, ok := mc.TrafficServer[string(result.ID)]
+ if !ok {
+ log.Errorf("Cache %v missing from from Traffic Ops Monitor Config - treating as OFFLINE\n", result.ID)
+ return false, "ERROR - server missing in Traffic Ops monitor config", ""
+ }
+ serverProfile, ok := mc.Profile[serverInfo.Profile]
+ if !ok {
+ log.Errorf("Cache %v profile %v missing from from Traffic Ops Monitor Config - treating as OFFLINE\n", result.ID, serverInfo.Profile)
+ return false, "ERROR - server profile missing in Traffic Ops monitor config", ""
+ }
+
+ status := enum.CacheStatusFromString(serverInfo.Status)
+ if status == enum.CacheStatusInvalid {
+ log.Errorf("Cache %v got invalid status from Traffic Ops '%v' - treating as Reported\n", result.ID, serverInfo.Status)
+ }
+
+ availability := "available"
+ if !result.Available {
+ availability = "unavailable"
+ }
+
+ switch {
+ case status == enum.CacheStatusInvalid:
+ log.Errorf("Cache %v got invalid status from Traffic Ops '%v' - treating as OFFLINE\n", result.ID, serverInfo.Status)
+ return false, getEventDescription(status, availability+"; invalid status"), ""
+ case status == enum.CacheStatusAdminDown:
+ return false, getEventDescription(status, availability), ""
+ case status == enum.CacheStatusOffline:
+ log.Errorf("Cache %v set to offline, but still polled\n", result.ID)
+ return false, getEventDescription(status, availability), ""
+ case status == enum.CacheStatusOnline:
+ return true, getEventDescription(status, availability), ""
+ case result.Error != nil:
+ return false, getEventDescription(status, fmt.Sprintf("%v", result.Error)), ""
+ }
+
+ computedStats := cache.ComputedStats()
+
+ for stat, threshold := range serverProfile.Parameters.Thresholds {
+ resultStat := interface{}(nil)
+ if computedStatF, ok := computedStats[stat]; ok {
+ dummyCombinedstate := peer.IsAvailable{} // the only stats which use combinedState are things like isAvailable, which don't make sense to ever be thresholds.
+ resultStat = computedStatF(result, serverInfo, serverProfile, dummyCombinedstate)
+ } else {
+ if resultStats == nil {
+ continue
+ }
+ resultStatHistory, ok := resultStats[stat]
+ if !ok {
+ continue
+ }
+ if len(resultStatHistory) < 1 {
+ continue
+ }
+ resultStat = resultStatHistory[0].Val
+ }
+
+ resultStatNum, ok := enum.ToNumeric(resultStat)
+ if !ok {
+ log.Errorf("health.EvalCache threshold stat %s was not a number: %v", stat, resultStat)
+ continue
+ }
+
+ if !InThreshold(threshold, resultStatNum) {
+ return false, getEventDescription(status, ExceedsThresholdMsg(stat, threshold, resultStatNum)), stat
+ }
+ }
+
+ return result.Available, getEventDescription(status, availability), ""
+}
+
+// ExceedsThresholdMsg returns a human-readable message for why the given value exceeds the threshold. It does NOT check whether the value actually exceeds the threshold; call `InThreshold` to check first.
+func ExceedsThresholdMsg(stat string, threshold to.HealthThreshold, val float64) string {
+ switch threshold.Comparator {
+ case "=":
+ return fmt.Sprintf("%s not equal (%f != %f)", stat, val, threshold.Val)
+ case ">":
+ return fmt.Sprintf("%s too low (%f < %f)", stat, val, threshold.Val)
+ case "<":
+ return fmt.Sprintf("%s too high (%f > %f)", stat, val, threshold.Val)
+ case ">=":
+ return fmt.Sprintf("%s too low (%f <= %f)", stat, val, threshold.Val)
+ case "<=":
+ return fmt.Sprintf("%s too high (%f >= %f)", stat, val, threshold.Val)
+ default:
+ return fmt.Sprintf("ERROR: Invalid Threshold: %+v", threshold)
+ }
+}
+
+func InThreshold(threshold to.HealthThreshold, val float64) bool {
+ switch threshold.Comparator {
+ case "=":
+ return val == threshold.Val
+ case ">":
+ return val > threshold.Val
+ case "<":
+ return val < threshold.Val
+ case ">=":
+ return val >= threshold.Val
+ case "<=":
+ return val <= threshold.Val
+ default:
+ log.Errorf("Invalid Threshold: %+v", threshold)
+ return true // for safety, if a threshold somehow gets corrupted, don't start marking caches down.
+ }
+}
+
+func getEventDescription(status enum.CacheStatus, message string) string {
+ return fmt.Sprintf("%s - %s", status, message)
+}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/3db9dd7b/traffic_monitor/experimental/traffic_monitor/health/cache_health.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/health/cache_health.go b/traffic_monitor/experimental/traffic_monitor/health/cache_health.go
deleted file mode 100644
index 0a00838..0000000
--- a/traffic_monitor/experimental/traffic_monitor/health/cache_health.go
+++ /dev/null
@@ -1,210 +0,0 @@
-package health
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-import (
- "fmt"
- "strconv"
- "strings"
-
- "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/common/log"
- "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/cache"
- "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/enum"
- "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/peer"
- to "github.com/apache/incubator-trafficcontrol/traffic_ops/client"
-)
-
-func setError(newResult *cache.Result, err error) {
- newResult.Error = err
- newResult.Available = false
-}
-
-// GetVitals Gets the vitals to decide health on in the right format
-func GetVitals(newResult *cache.Result, prevResult *cache.Result, mc *to.TrafficMonitorConfigMap) {
- if newResult.Error != nil {
- log.Errorf("cache_health.GetVitals() called with an errored Result!")
- return
- }
- // proc.loadavg -- we're using the 1 minute average (!?)
- // value looks like: "0.20 0.07 0.07 1/967 29536" (without the quotes)
- loadAverages := strings.Fields(newResult.Astats.System.ProcLoadavg)
- if len(loadAverages) > 0 {
- oneMinAvg, err := strconv.ParseFloat(loadAverages[0], 64)
- if err != nil {
- setError(newResult, fmt.Errorf("Error converting load average string '%s': %v", newResult.Astats.System.ProcLoadavg, err))
- return
- }
- newResult.Vitals.LoadAvg = oneMinAvg
- } else {
- setError(newResult, fmt.Errorf("Can't make sense of '%s' as a load average for %s", newResult.Astats.System.ProcLoadavg, newResult.ID))
- return
- }
-
- // proc.net.dev -- need to compare to prevSample
- // value looks like
- // "bond0:8495786321839 31960528603 0 0 0 0 0 2349716 143283576747316 101104535041 0 0 0 0 0 0"
- // (without the quotes)
- parts := strings.Split(newResult.Astats.System.ProcNetDev, ":")
- if len(parts) > 1 {
- numbers := strings.Fields(parts[1])
- var err error
- newResult.Vitals.BytesOut, err = strconv.ParseInt(numbers[8], 10, 64)
- if err != nil {
- setError(newResult, fmt.Errorf("Error converting BytesOut from procnetdev: %v", err))
- return
- }
- newResult.Vitals.BytesIn, err = strconv.ParseInt(numbers[0], 10, 64)
- if err != nil {
- setError(newResult, fmt.Errorf("Error converting BytesIn from procnetdev: %v", err))
- return
- }
- if prevResult != nil && prevResult.Vitals.BytesOut != 0 {
- elapsedTimeInSecs := float64(newResult.Time.UnixNano()-prevResult.Time.UnixNano()) / 1000000000
- newResult.Vitals.KbpsOut = int64(float64(((newResult.Vitals.BytesOut - prevResult.Vitals.BytesOut) * 8 / 1000)) / elapsedTimeInSecs)
- } else {
- // log.Infoln("prevResult == nil for id " + newResult.Id + ". Hope we're just starting up?")
- }
- } else {
- setError(newResult, fmt.Errorf("Error parsing procnetdev: no fields found"))
- return
- }
-
- // inf.speed -- value looks like "10000" (without the quotes) so it is in Mbps.
- // TODO JvD: Should we really be running this code every second for every cache polled????? I don't think so.
- interfaceBandwidth := newResult.Astats.System.InfSpeed
- newResult.Vitals.MaxKbpsOut = int64(interfaceBandwidth) * 1000
-
- // log.Infoln(newResult.Id, "BytesOut", newResult.Vitals.BytesOut, "BytesIn", newResult.Vitals.BytesIn, "Kbps", newResult.Vitals.KbpsOut, "max", newResult.Vitals.MaxKbpsOut)
-}
-
-// EvalCache returns whether the given cache should be marked available, a string describing why, and which stat exceeded a threshold. The `stats` may be nil, for pollers which don't poll stats.
-// The availability of EvalCache MAY NOT be used to directly set the cache's local availability, because the threshold stats may not be part of the poller which produced the result. Rather, if the cache was previously unavailable from a threshold, it must be verified that threshold stat is in the results before setting the cache to available.
-// TODO change to return a `cache.AvailableStatus`
-func EvalCache(result cache.ResultInfo, resultStats cache.ResultStatValHistory, mc *to.TrafficMonitorConfigMap) (bool, string, string) {
- serverInfo, ok := mc.TrafficServer[string(result.ID)]
- if !ok {
- log.Errorf("Cache %v missing from from Traffic Ops Monitor Config - treating as OFFLINE\n", result.ID)
- return false, "ERROR - server missing in Traffic Ops monitor config", ""
- }
- serverProfile, ok := mc.Profile[serverInfo.Profile]
- if !ok {
- log.Errorf("Cache %v profile %v missing from from Traffic Ops Monitor Config - treating as OFFLINE\n", result.ID, serverInfo.Profile)
- return false, "ERROR - server profile missing in Traffic Ops monitor config", ""
- }
-
- status := enum.CacheStatusFromString(serverInfo.Status)
- if status == enum.CacheStatusInvalid {
- log.Errorf("Cache %v got invalid status from Traffic Ops '%v' - treating as Reported\n", result.ID, serverInfo.Status)
- }
-
- availability := "available"
- if !result.Available {
- availability = "unavailable"
- }
-
- switch {
- case status == enum.CacheStatusInvalid:
- log.Errorf("Cache %v got invalid status from Traffic Ops '%v' - treating as OFFLINE\n", result.ID, serverInfo.Status)
- return false, getEventDescription(status, availability+"; invalid status"), ""
- case status == enum.CacheStatusAdminDown:
- return false, getEventDescription(status, availability), ""
- case status == enum.CacheStatusOffline:
- log.Errorf("Cache %v set to offline, but still polled\n", result.ID)
- return false, getEventDescription(status, availability), ""
- case status == enum.CacheStatusOnline:
- return true, getEventDescription(status, availability), ""
- case result.Error != nil:
- return false, getEventDescription(status, fmt.Sprintf("%v", result.Error)), ""
- }
-
- computedStats := cache.ComputedStats()
-
- for stat, threshold := range serverProfile.Parameters.Thresholds {
- resultStat := interface{}(nil)
- if computedStatF, ok := computedStats[stat]; ok {
- dummyCombinedstate := peer.IsAvailable{} // the only stats which use combinedState are things like isAvailable, which don't make sense to ever be thresholds.
- resultStat = computedStatF(result, serverInfo, serverProfile, dummyCombinedstate)
- } else {
- if resultStats == nil {
- continue
- }
- resultStatHistory, ok := resultStats[stat]
- if !ok {
- continue
- }
- if len(resultStatHistory) < 1 {
- continue
- }
- resultStat = resultStatHistory[0].Val
- }
-
- resultStatNum, ok := enum.ToNumeric(resultStat)
- if !ok {
- log.Errorf("health.EvalCache threshold stat %s was not a number: %v", stat, resultStat)
- continue
- }
-
- if !InThreshold(threshold, resultStatNum) {
- return false, getEventDescription(status, ExceedsThresholdMsg(stat, threshold, resultStatNum)), stat
- }
- }
-
- return result.Available, getEventDescription(status, availability), ""
-}
-
-// ExceedsThresholdMsg returns a human-readable message for why the given value exceeds the threshold. It does NOT check whether the value actually exceeds the threshold; call `InThreshold` to check first.
-func ExceedsThresholdMsg(stat string, threshold to.HealthThreshold, val float64) string {
- switch threshold.Comparator {
- case "=":
- return fmt.Sprintf("%s not equal (%f != %f)", stat, val, threshold.Val)
- case ">":
- return fmt.Sprintf("%s too low (%f < %f)", stat, val, threshold.Val)
- case "<":
- return fmt.Sprintf("%s too high (%f > %f)", stat, val, threshold.Val)
- case ">=":
- return fmt.Sprintf("%s too low (%f <= %f)", stat, val, threshold.Val)
- case "<=":
- return fmt.Sprintf("%s too high (%f >= %f)", stat, val, threshold.Val)
- default:
- return fmt.Sprintf("ERROR: Invalid Threshold: %+v", threshold)
- }
-}
-
-func InThreshold(threshold to.HealthThreshold, val float64) bool {
- switch threshold.Comparator {
- case "=":
- return val == threshold.Val
- case ">":
- return val > threshold.Val
- case "<":
- return val < threshold.Val
- case ">=":
- return val >= threshold.Val
- case "<=":
- return val <= threshold.Val
- default:
- log.Errorf("Invalid Threshold: %+v", threshold)
- return true // for safety, if a threshold somehow gets corrupted, don't start marking caches down.
- }
-}
-
-func getEventDescription(status enum.CacheStatus, message string) string {
- return fmt.Sprintf("%s - %s", status, message)
-}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/3db9dd7b/traffic_monitor/experimental/traffic_monitor/manager/stathistory.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/stathistory.go b/traffic_monitor/experimental/traffic_monitor/manager/stathistory.go
index c63584e..a4a9b81 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/stathistory.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/stathistory.go
@@ -255,7 +255,7 @@ func calcAvailability(results []cache.Result, pollerName string, statResultHisto
if available, ok := localStates.GetCache(result.ID); !ok || available.IsAvailable != isAvailable {
log.Infof("Changing state for %s was: %t now: %t because %s poller: %v error: %v", result.ID, available.IsAvailable, isAvailable, whyAvailable, pollerName, result.Error)
- events.Add(cache.Event{Time: time.Now().Unix(), Description: whyAvailable + " (" + pollerName + ")", Name: result.ID, Hostname: result.ID, Type: toData.ServerTypes[result.ID].String(), Available: isAvailable})
+ events.Add(health.Event{Time: time.Now(), Description: whyAvailable + " (" + pollerName + ")", Name: string(result.ID), Hostname: string(result.ID), Type: toData.ServerTypes[result.ID].String(), Available: isAvailable})
}
localStates.SetCache(result.ID, peer.IsAvailable{IsAvailable: isAvailable})
[09/20] incubator-trafficcontrol git commit: Remove TM2 wrong comments
Posted by ne...@apache.org.
Remove TM2 wrong comments
Project: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/commit/f3e45f79
Tree: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/tree/f3e45f79
Diff: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/diff/f3e45f79
Branch: refs/heads/master
Commit: f3e45f79c1831f6004d56b427e0c4ca160b06356
Parents: f31ffc7
Author: Robert Butts <ro...@gmail.com>
Authored: Mon Jan 23 11:02:38 2017 -0700
Committer: Dave Neuman <ne...@apache.org>
Committed: Wed Jan 25 10:29:46 2017 -0700
----------------------------------------------------------------------
.../experimental/traffic_monitor/manager/datarequest.go | 4 ++--
.../experimental/traffic_monitor/manager/manager.go | 3 +--
.../traffic_monitor/trafficopsdata/trafficopsdata.go | 6 ++----
3 files changed, 5 insertions(+), 8 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/f3e45f79/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go b/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
index 2dffb41..47853b7 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
@@ -584,7 +584,6 @@ func srvTRStateSelf(localStates peer.CRStatesThreadsafe) ([]byte, error) {
return peer.CrstatesMarshall(localStates.Get())
}
-// TODO remove error params, handle by returning an error? How, since we need to return a non-standard code?
func srvCacheStats(params url.Values, errorCount threadsafe.Uint, path string, toData todata.TODataThreadsafe, statResultHistory threadsafe.ResultStatHistory, statInfoHistory threadsafe.ResultInfoHistory, monitorConfig TrafficMonitorConfigMapThreadsafe, combinedStates peer.CRStatesThreadsafe, statMaxKbpses threadsafe.CacheKbpses) ([]byte, int) {
filter, err := NewCacheStatFilter(path, params, toData.Get().ServerTypes)
if err != nil {
@@ -1200,7 +1199,8 @@ func createStatSummary(statResultHistory cache.ResultStatHistory, filter cache.F
for _, val := range statHistory {
fVal, ok := enum.ToNumeric(val.Val)
if !ok {
- continue // skip non-numeric stats. TODO warn about stat history containing different types?
+ log.Warnf("threshold stat %v value %v is not a number, cannot use.", statName, val.Val)
+ continue
}
for i := uint64(0); i < val.Span; i++ {
ssStat.DataPointCount++
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/f3e45f79/traffic_monitor/experimental/traffic_monitor/manager/manager.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/manager.go b/traffic_monitor/experimental/traffic_monitor/manager/manager.go
index b5e3d49..79c69af 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/manager.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/manager.go
@@ -60,7 +60,6 @@ func Start(opsConfigFile string, cfg config.Config, staticAppData StaticAppData)
Pending: gmx.NewGauge("fetchPending"),
}
- // TODO investigate whether a unique client per cache to be polled is faster
sharedClient := &http.Client{
Transport: &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true}},
Timeout: cfg.HTTPTimeout,
@@ -76,7 +75,7 @@ func Start(opsConfigFile string, cfg config.Config, staticAppData StaticAppData)
cacheHealthHandler := cache.NewHandler()
cacheHealthPoller := poller.NewHTTP(cfg.CacheHealthPollingInterval, true, sharedClient, counters, cacheHealthHandler, cfg.HTTPPollNoSleep)
- cacheStatHandler := cache.NewPrecomputeHandler(toData, peerStates) // TODO figure out if this is necessary, with the CacheHealthPoller
+ cacheStatHandler := cache.NewPrecomputeHandler(toData, peerStates)
cacheStatPoller := poller.NewHTTP(cfg.CacheStatPollingInterval, false, sharedClient, counters, cacheStatHandler, cfg.HTTPPollNoSleep)
monitorConfigPoller := poller.NewMonitorConfig(cfg.MonitorConfigPollingInterval)
peerHandler := peer.NewHandler()
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/f3e45f79/traffic_monitor/experimental/traffic_monitor/trafficopsdata/trafficopsdata.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/trafficopsdata/trafficopsdata.go b/traffic_monitor/experimental/traffic_monitor/trafficopsdata/trafficopsdata.go
index e12818c..cefb441 100644
--- a/traffic_monitor/experimental/traffic_monitor/trafficopsdata/trafficopsdata.go
+++ b/traffic_monitor/experimental/traffic_monitor/trafficopsdata/trafficopsdata.go
@@ -8,9 +8,9 @@ package trafficopsdata
* to you under the Apache License, Version 2.0 (the
* "License"); you may not use this file except in compliance
* with the License. You may obtain a copy of the License at
- *
+ *
* http://www.apache.org/licenses/LICENSE-2.0
- *
+ *
* Unless required by applicable law or agreed to in writing,
* software distributed under the License is distributed on an
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
@@ -19,7 +19,6 @@ package trafficopsdata
* under the License.
*/
-
import (
"encoding/json"
"fmt"
@@ -134,7 +133,6 @@ type CRConfig struct {
// Fetch gets the CRConfig from Traffic Ops, creates the TOData maps, and atomically sets the TOData.
// TODO since the session is threadsafe, each TOData get func below could be put in a goroutine, if performance mattered
-// TODO change called funcs to take CRConfigRaw instead of toSession
func (d TODataThreadsafe) Fetch(to towrap.ITrafficOpsSession, cdn string) error {
newTOData := TOData{}
[05/20] incubator-trafficcontrol git commit: Add TM2 cache capacity
to GUI
Posted by ne...@apache.org.
Add TM2 cache capacity to GUI
Project: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/commit/9c6dbea1
Tree: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/tree/9c6dbea1
Diff: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/diff/9c6dbea1
Branch: refs/heads/master
Commit: 9c6dbea102b90d3a1686132556577bd317394814
Parents: cef0859
Author: Robert Butts <ro...@gmail.com>
Authored: Wed Jan 18 08:45:20 2017 -0700
Committer: Dave Neuman <ne...@apache.org>
Committed: Wed Jan 25 10:29:46 2017 -0700
----------------------------------------------------------------------
traffic_monitor/experimental/traffic_monitor/static/index.html | 1 -
1 file changed, 1 deletion(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/9c6dbea1/traffic_monitor/experimental/traffic_monitor/static/index.html
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/static/index.html b/traffic_monitor/experimental/traffic_monitor/static/index.html
index 0ed337e..66cd03b 100644
--- a/traffic_monitor/experimental/traffic_monitor/static/index.html
+++ b/traffic_monitor/experimental/traffic_monitor/static/index.html
@@ -324,7 +324,6 @@ under the License.
var kbps = (jdata[server].bandwidth_kbps / kilobitsInMegabit).toFixed(2);
var max = numberStrWithCommas((jdata[server].bandwidth_capacity_kbps / kilobitsInMegabit).toFixed(0));
document.getElementById("cache-states-" + server + "-bandwidth").textContent = '' + kbps + ' / ' + max;
-
} else {
document.getElementById("cache-states-" + server + "-bandwidth").textContent = "N/A";
}
[20/20] incubator-trafficcontrol git commit: This closes #204
Posted by ne...@apache.org.
This closes #204
Project: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/commit/246e2e63
Tree: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/tree/246e2e63
Diff: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/diff/246e2e63
Branch: refs/heads/master
Commit: 246e2e633718e52c6ee57a622c110c60dbb7ed8c
Parents: 091b3c9
Author: Dave Neuman <ne...@apache.org>
Authored: Wed Jan 25 10:30:01 2017 -0700
Committer: Dave Neuman <ne...@apache.org>
Committed: Wed Jan 25 10:30:01 2017 -0700
----------------------------------------------------------------------
----------------------------------------------------------------------
[07/20] incubator-trafficcontrol git commit: Fix TM2 duplicate
availability logic
Posted by ne...@apache.org.
Fix TM2 duplicate availability logic
Project: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/commit/40d5bcba
Tree: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/tree/40d5bcba
Diff: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/diff/40d5bcba
Branch: refs/heads/master
Commit: 40d5bcbaa78a1459c57d8a05fb945817d58323ec
Parents: 8a7bed4
Author: Robert Butts <ro...@gmail.com>
Authored: Fri Jan 20 15:30:35 2017 -0700
Committer: Dave Neuman <ne...@apache.org>
Committed: Wed Jan 25 10:29:46 2017 -0700
----------------------------------------------------------------------
.../traffic_monitor/manager/healthresult.go | 30 ++--------
.../traffic_monitor/manager/stathistory.go | 59 +++++++++++---------
2 files changed, 38 insertions(+), 51 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/40d5bcba/traffic_monitor/experimental/traffic_monitor/manager/healthresult.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/healthresult.go b/traffic_monitor/experimental/traffic_monitor/manager/healthresult.go
index f99155d..110f083 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/healthresult.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/healthresult.go
@@ -204,10 +204,9 @@ func processHealthResult(
}()
toDataCopy := toData.Get() // create a copy, so the same data used for all processing of this cache health result
- localCacheStatus := localCacheStatusThreadsafe.Get().Copy()
monitorConfigCopy := monitorConfig.Get()
healthHistoryCopy := healthHistory.Get().Copy()
- for _, healthResult := range results {
+ for i, healthResult := range results {
log.Debugf("poll %v %v healthresultman start\n", healthResult.PollID, time.Now())
fetchCount.Inc()
var prevResult cache.Result
@@ -218,6 +217,7 @@ func processHealthResult(
if healthResult.Error == nil {
health.GetVitals(&healthResult, &prevResult, &monitorConfigCopy)
+ results[i] = healthResult
}
maxHistory := uint64(monitorConfigCopy.Profile[monitorConfigCopy.TrafficServer[string(healthResult.ID)].Profile].Parameters.HistoryCount)
@@ -227,33 +227,11 @@ func processHealthResult(
}
healthHistoryCopy[healthResult.ID] = pruneHistory(append([]cache.Result{healthResult}, healthHistoryCopy[healthResult.ID]...), maxHistory)
+ }
- isAvailable, whyAvailable, unavailableStat := health.EvalCache(cache.ToInfo(healthResult), nil, &monitorConfigCopy)
- whyAvailable += "(healthpoll)" // debug
- if available, ok := localStates.GetCache(healthResult.ID); !ok || available.IsAvailable != isAvailable {
- log.Infof("Changing state for %s was: %t now: %t because %s error: %v", healthResult.ID, available.IsAvailable, isAvailable, whyAvailable, healthResult.Error)
- events.Add(health.Event{Time: time.Now(), Description: whyAvailable, Name: string(healthResult.ID), Hostname: string(healthResult.ID), Type: toDataCopy.ServerTypes[healthResult.ID].String(), Available: isAvailable})
- }
+ calcAvailability(results, "health", nil, monitorConfigCopy, toDataCopy, localCacheStatusThreadsafe, localStates, events)
- // if the cache is now Available, and was previously unavailable due to a threshold, make sure this poller contains the stat which exceeded the threshold.
- if previousStatus, hasPreviousStatus := localCacheStatus[healthResult.ID]; isAvailable && hasPreviousStatus && !previousStatus.Available && previousStatus.UnavailableStat != "" {
- if !resultHasStat(previousStatus.UnavailableStat, healthResult) {
- // TODO determine if it's ok to add the result data (but not availability). Or will making them not align cause issues?
- continue
- }
- }
-
- localCacheStatus[healthResult.ID] = cache.AvailableStatus{
- Available: isAvailable,
- Status: monitorConfigCopy.TrafficServer[string(healthResult.ID)].Status,
- Why: whyAvailable,
- UnavailableStat: unavailableStat,
- } // TODO move within localStates?
- localStates.SetCache(healthResult.ID, peer.IsAvailable{IsAvailable: isAvailable})
- }
- CalculateDeliveryServiceState(toDataCopy.DeliveryServiceServers, localStates)
healthHistory.Set(healthHistoryCopy)
- localCacheStatusThreadsafe.Set(localCacheStatus)
// TODO determine if we should combineCrStates() here
lastHealthDurations := lastHealthDurationsThreadsafe.Get().Copy()
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/40d5bcba/traffic_monitor/experimental/traffic_monitor/manager/stathistory.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/stathistory.go b/traffic_monitor/experimental/traffic_monitor/manager/stathistory.go
index bf2d1dc..b44f357 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/stathistory.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/stathistory.go
@@ -158,7 +158,6 @@ func processStatResults(
statInfoHistory := statInfoHistoryThreadsafe.Get().Copy()
statResultHistory := statResultHistoryThreadsafe.Get().Copy()
statMaxKbpses := statMaxKbpsesThreadsafe.Get().Copy()
- localCacheStatus := localCacheStatusThreadsafe.Get().Copy()
for i, result := range results {
maxStats := uint64(mc.Profile[mc.TrafficServer[string(result.ID)].Profile].Parameters.HistoryCount)
@@ -213,44 +212,54 @@ func processStatResults(
lastStats.Set(newLastStats)
}
- // TODO test
- // TODO abstract setting availability logic (duplicated in healthresult.go)
+ calcAvailability(results, "stat", statResultHistory, mc, toData, localCacheStatusThreadsafe, localStates, events)
+
+ endTime := time.Now()
+ lastStatDurations := lastStatDurationsThreadsafe.Get().Copy()
for _, result := range results {
- isAvailable, whyAvailable, unavailableStat := health.EvalCache(cache.ToInfo(result), statResultHistory[result.ID], &mc)
- whyAvailable += "(statpoll)" // debug
+ if lastStatStart, ok := lastStatEndTimes[result.ID]; ok {
+ d := time.Since(lastStatStart)
+ lastStatDurations[result.ID] = d
+ }
+ lastStatEndTimes[result.ID] = endTime
+ }
+ lastStatDurationsThreadsafe.Set(lastStatDurations)
+ unpolledCaches.SetPolled(results, lastStats.Get())
+}
- if available, ok := localStates.GetCache(result.ID); !ok || available.IsAvailable != isAvailable {
- log.Infof("Changing state for %s was: %t now: %t because %s error: %v", result.ID, available.IsAvailable, isAvailable, whyAvailable, result.Error)
- events.Add(health.Event{Time: time.Now(), Description: whyAvailable, Name: string(result.ID), Hostname: string(result.ID), Type: toData.ServerTypes[result.ID].String(), Available: isAvailable})
+// calcAvailability calculates the availability of the cache, from the given result. Availability is stored in `localCacheStatus` and `localStates`, and if the status changed an event is added to `events`. statResultHistory may be nil, for pollers which don't poll stats.
+// TODO add enum for poller names?
+func calcAvailability(results []cache.Result, pollerName string, statResultHistory cache.ResultStatHistory, mc to.TrafficMonitorConfigMap, toData todata.TOData, localCacheStatusThreadsafe threadsafe.CacheAvailableStatus, localStates peer.CRStatesThreadsafe, events threadsafe.Events) {
+ localCacheStatuses := localCacheStatusThreadsafe.Get().Copy()
+ for _, result := range results {
+ statResults := cache.ResultStatValHistory(nil)
+ if statResultHistory != nil {
+ statResults = statResultHistory[result.ID]
}
+ isAvailable, whyAvailable, unavailableStat := health.EvalCache(cache.ToInfo(result), statResults, &mc)
+ whyAvailable += " (" + pollerName + ")" // TODO move to field in AvailableStatus
+
// if the cache is now Available, and was previously unavailable due to a threshold, make sure this poller contains the stat which exceeded the threshold.
- if previousStatus, hasPreviousStatus := localCacheStatus[result.ID]; isAvailable && hasPreviousStatus && !previousStatus.Available && previousStatus.UnavailableStat != "" {
+ if previousStatus, hasPreviousStatus := localCacheStatuses[result.ID]; isAvailable && hasPreviousStatus && !previousStatus.Available && previousStatus.UnavailableStat != "" {
if !resultHasStat(previousStatus.UnavailableStat, result) {
- // TODO determine if it's ok to add the result data (but not availability). Or will making them not align cause issues?
- continue
+ return
}
}
- localCacheStatus[result.ID] = cache.AvailableStatus{
+ localCacheStatuses[result.ID] = cache.AvailableStatus{
Available: isAvailable,
Status: mc.TrafficServer[string(result.ID)].Status,
Why: whyAvailable,
UnavailableStat: unavailableStat,
} // TODO move within localStates?
- localStates.SetCache(result.ID, peer.IsAvailable{IsAvailable: isAvailable})
- CalculateDeliveryServiceState(toData.DeliveryServiceServers, localStates)
- localCacheStatusThreadsafe.Set(localCacheStatus)
- }
- endTime := time.Now()
- lastStatDurations := lastStatDurationsThreadsafe.Get().Copy()
- for _, result := range results {
- if lastStatStart, ok := lastStatEndTimes[result.ID]; ok {
- d := time.Since(lastStatStart)
- lastStatDurations[result.ID] = d
+ if available, ok := localStates.GetCache(result.ID); !ok || available.IsAvailable != isAvailable {
+ log.Infof("Changing state for %s was: %t now: %t because %s error: %v", result.ID, available.IsAvailable, isAvailable, whyAvailable, result.Error)
+ events.Add(cache.Event{Time: time.Now().Unix(), Description: whyAvailable, Name: result.ID, Hostname: result.ID, Type: toData.ServerTypes[result.ID].String(), Available: isAvailable})
}
- lastStatEndTimes[result.ID] = endTime
+
+ localStates.SetCache(result.ID, peer.IsAvailable{IsAvailable: isAvailable})
}
- lastStatDurationsThreadsafe.Set(lastStatDurations)
- unpolledCaches.SetPolled(results, lastStats.Get())
+ CalculateDeliveryServiceState(toData.DeliveryServiceServers, localStates)
+ localCacheStatusThreadsafe.Set(localCacheStatuses)
}
[17/20] incubator-trafficcontrol git commit: Move TM2 ToNumeric() to
util package
Posted by ne...@apache.org.
Move TM2 ToNumeric() to util package
Project: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/commit/f9115dcc
Tree: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/tree/f9115dcc
Diff: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/diff/f9115dcc
Branch: refs/heads/master
Commit: f9115dcc1a3bd98000f8f94ade51aaefdcd3ed50
Parents: dea90b1
Author: Robert Butts <ro...@gmail.com>
Authored: Tue Jan 24 14:07:57 2017 -0700
Committer: Dave Neuman <ne...@apache.org>
Committed: Wed Jan 25 10:29:46 2017 -0700
----------------------------------------------------------------------
traffic_monitor/experimental/common/util/num.go | 35 ++++++++++++++++++++
.../experimental/traffic_monitor/enum/enum.go | 35 --------------------
.../traffic_monitor/health/cache.go | 3 +-
.../traffic_monitor/manager/datarequest.go | 7 ++--
4 files changed, 41 insertions(+), 39 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/f9115dcc/traffic_monitor/experimental/common/util/num.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/common/util/num.go b/traffic_monitor/experimental/common/util/num.go
new file mode 100644
index 0000000..4db2cf9
--- /dev/null
+++ b/traffic_monitor/experimental/common/util/num.go
@@ -0,0 +1,35 @@
+package util
+
+// ToNumeric returns a float for any numeric type, and false if the interface does not hold a numeric type.
+// This allows converting unknown numeric types (for example, from JSON) in a single line
+// TODO try to parse string stats as numbers?
+func ToNumeric(v interface{}) (float64, bool) {
+ switch i := v.(type) {
+ case uint8:
+ return float64(i), true
+ case uint16:
+ return float64(i), true
+ case uint32:
+ return float64(i), true
+ case uint64:
+ return float64(i), true
+ case int8:
+ return float64(i), true
+ case int16:
+ return float64(i), true
+ case int32:
+ return float64(i), true
+ case int64:
+ return float64(i), true
+ case float32:
+ return float64(i), true
+ case float64:
+ return i, true
+ case int:
+ return float64(i), true
+ case uint:
+ return float64(i), true
+ default:
+ return 0.0, false
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/f9115dcc/traffic_monitor/experimental/traffic_monitor/enum/enum.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/enum/enum.go b/traffic_monitor/experimental/traffic_monitor/enum/enum.go
index 0e43016..853236a 100644
--- a/traffic_monitor/experimental/traffic_monitor/enum/enum.go
+++ b/traffic_monitor/experimental/traffic_monitor/enum/enum.go
@@ -175,38 +175,3 @@ func CacheStatusFromString(s string) CacheStatus {
return CacheStatusInvalid
}
}
-
-// toNumeric returns a float for any numeric type, and false if the interface does not hold a numeric type.
-// This allows converting unknown numeric types (for example, from JSON) in a single line
-// TODO move to 'util' or 'tools' package?
-// TODO try to parse string stats as numbers?
-func ToNumeric(v interface{}) (float64, bool) {
- switch i := v.(type) {
- case uint8:
- return float64(i), true
- case uint16:
- return float64(i), true
- case uint32:
- return float64(i), true
- case uint64:
- return float64(i), true
- case int8:
- return float64(i), true
- case int16:
- return float64(i), true
- case int32:
- return float64(i), true
- case int64:
- return float64(i), true
- case float32:
- return float64(i), true
- case float64:
- return i, true
- case int:
- return float64(i), true
- case uint:
- return float64(i), true
- default:
- return 0.0, false
- }
-}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/f9115dcc/traffic_monitor/experimental/traffic_monitor/health/cache.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/health/cache.go b/traffic_monitor/experimental/traffic_monitor/health/cache.go
index 533652a..fd3fcec 100644
--- a/traffic_monitor/experimental/traffic_monitor/health/cache.go
+++ b/traffic_monitor/experimental/traffic_monitor/health/cache.go
@@ -26,6 +26,7 @@ import (
"time"
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/common/log"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/common/util"
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/cache"
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/enum"
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/peer"
@@ -153,7 +154,7 @@ func EvalCache(result cache.ResultInfo, resultStats cache.ResultStatValHistory,
resultStat = resultStatHistory[0].Val
}
- resultStatNum, ok := enum.ToNumeric(resultStat)
+ resultStatNum, ok := util.ToNumeric(resultStat)
if !ok {
log.Errorf("health.EvalCache threshold stat %s was not a number: %v", stat, resultStat)
continue
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/f9115dcc/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go b/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
index eddf972..7712154 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
@@ -31,6 +31,7 @@ import (
"time"
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/common/log"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/common/util"
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/cache"
ds "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/deliveryservice"
dsdata "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/deliveryservicedata"
@@ -1175,8 +1176,8 @@ func createStatSummary(statResultHistory cache.ResultStatHistory, filter cache.F
msPerNs := int64(1000000)
ssStat.StartTime = statHistory[len(statHistory)-1].Time.UnixNano() / msPerNs
ssStat.EndTime = statHistory[0].Time.UnixNano() / msPerNs
- oldestVal, isOldestValNumeric := enum.ToNumeric(statHistory[len(statHistory)-1].Val)
- newestVal, isNewestValNumeric := enum.ToNumeric(statHistory[0].Val)
+ oldestVal, isOldestValNumeric := util.ToNumeric(statHistory[len(statHistory)-1].Val)
+ newestVal, isNewestValNumeric := util.ToNumeric(statHistory[0].Val)
if !isOldestValNumeric || !isNewestValNumeric {
continue // skip non-numeric stats
}
@@ -1185,7 +1186,7 @@ func createStatSummary(statResultHistory cache.ResultStatHistory, filter cache.F
ssStat.High = newestVal
ssStat.Low = newestVal
for _, val := range statHistory {
- fVal, ok := enum.ToNumeric(val.Val)
+ fVal, ok := util.ToNumeric(val.Val)
if !ok {
log.Warnf("threshold stat %v value %v is not a number, cannot use.", statName, val.Val)
continue
[15/20] incubator-trafficcontrol git commit: Rename TM2 setError to
setErr
Posted by ne...@apache.org.
Rename TM2 setError to setErr
Project: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/commit/cc10a02f
Tree: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/tree/cc10a02f
Diff: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/diff/cc10a02f
Branch: refs/heads/master
Commit: cc10a02faed08d8804438cce1be4ff2e116c5ab8
Parents: 3db9dd7
Author: Robert Butts <ro...@gmail.com>
Authored: Tue Jan 24 10:44:41 2017 -0700
Committer: Dave Neuman <ne...@apache.org>
Committed: Wed Jan 25 10:29:46 2017 -0700
----------------------------------------------------------------------
.../experimental/traffic_monitor/health/cache.go | 12 ++++++------
1 file changed, 6 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/cc10a02f/traffic_monitor/experimental/traffic_monitor/health/cache.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/health/cache.go b/traffic_monitor/experimental/traffic_monitor/health/cache.go
index 0a00838..68f64a7 100644
--- a/traffic_monitor/experimental/traffic_monitor/health/cache.go
+++ b/traffic_monitor/experimental/traffic_monitor/health/cache.go
@@ -31,7 +31,7 @@ import (
to "github.com/apache/incubator-trafficcontrol/traffic_ops/client"
)
-func setError(newResult *cache.Result, err error) {
+func setErr(newResult *cache.Result, err error) {
newResult.Error = err
newResult.Available = false
}
@@ -48,12 +48,12 @@ func GetVitals(newResult *cache.Result, prevResult *cache.Result, mc *to.Traffic
if len(loadAverages) > 0 {
oneMinAvg, err := strconv.ParseFloat(loadAverages[0], 64)
if err != nil {
- setError(newResult, fmt.Errorf("Error converting load average string '%s': %v", newResult.Astats.System.ProcLoadavg, err))
+ setErr(newResult, fmt.Errorf("Error converting load average string '%s': %v", newResult.Astats.System.ProcLoadavg, err))
return
}
newResult.Vitals.LoadAvg = oneMinAvg
} else {
- setError(newResult, fmt.Errorf("Can't make sense of '%s' as a load average for %s", newResult.Astats.System.ProcLoadavg, newResult.ID))
+ setErr(newResult, fmt.Errorf("Can't make sense of '%s' as a load average for %s", newResult.Astats.System.ProcLoadavg, newResult.ID))
return
}
@@ -67,12 +67,12 @@ func GetVitals(newResult *cache.Result, prevResult *cache.Result, mc *to.Traffic
var err error
newResult.Vitals.BytesOut, err = strconv.ParseInt(numbers[8], 10, 64)
if err != nil {
- setError(newResult, fmt.Errorf("Error converting BytesOut from procnetdev: %v", err))
+ setErr(newResult, fmt.Errorf("Error converting BytesOut from procnetdev: %v", err))
return
}
newResult.Vitals.BytesIn, err = strconv.ParseInt(numbers[0], 10, 64)
if err != nil {
- setError(newResult, fmt.Errorf("Error converting BytesIn from procnetdev: %v", err))
+ setErr(newResult, fmt.Errorf("Error converting BytesIn from procnetdev: %v", err))
return
}
if prevResult != nil && prevResult.Vitals.BytesOut != 0 {
@@ -82,7 +82,7 @@ func GetVitals(newResult *cache.Result, prevResult *cache.Result, mc *to.Traffic
// log.Infoln("prevResult == nil for id " + newResult.Id + ". Hope we're just starting up?")
}
} else {
- setError(newResult, fmt.Errorf("Error parsing procnetdev: no fields found"))
+ setErr(newResult, fmt.Errorf("Error parsing procnetdev: no fields found"))
return
}
[12/20] incubator-trafficcontrol git commit: Remove TM2 unused func
Posted by ne...@apache.org.
Remove TM2 unused func
Project: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/commit/6760764c
Tree: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/tree/6760764c
Diff: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/diff/6760764c
Branch: refs/heads/master
Commit: 6760764cadd9ad5f2a8226371533a739ec731dbc
Parents: f3e45f7
Author: Robert Butts <ro...@gmail.com>
Authored: Mon Jan 23 13:49:23 2017 -0700
Committer: Dave Neuman <ne...@apache.org>
Committed: Wed Jan 25 10:29:46 2017 -0700
----------------------------------------------------------------------
.../experimental/traffic_monitor/cache/cache.go | 20 --------------------
1 file changed, 20 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/6760764c/traffic_monitor/experimental/traffic_monitor/cache/cache.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/cache/cache.go b/traffic_monitor/experimental/traffic_monitor/cache/cache.go
index a850102..a0b6b60 100644
--- a/traffic_monitor/experimental/traffic_monitor/cache/cache.go
+++ b/traffic_monitor/experimental/traffic_monitor/cache/cache.go
@@ -280,26 +280,6 @@ func StatsMarshall(statResultHistory ResultStatHistory, statInfo ResultInfoHisto
return json.Marshal(stats)
}
-func getHealthPollingURL(cache enum.CacheName, monitorConfig to.TrafficMonitorConfigMap) (string, error) {
- srv, ok := monitorConfig.TrafficServer[string(cache)]
- if !ok {
- return "", fmt.Errorf("server not found in monitor config from Traffic Ops")
- }
- url := monitorConfig.Profile[srv.Profile].Parameters.HealthPollingURL
- if url == "" {
- return "", fmt.Errorf("health polling URL not found in monitor config from Traffic Ops")
- }
-
- // TODO abstract replacer, remove duplication with manager/monitorconfig.go
- url = strings.NewReplacer(
- "${hostname}", srv.IP,
- "${interface_name}", srv.InterfaceName,
- "application=system", "application=plugin.remap",
- "application=", "application=plugin.remap",
- ).Replace(url)
- return url, nil
-}
-
// Handle handles results fetched from a cache, parsing the raw Reader data and passing it along to a chan for further processing.
func (handler Handler) Handle(id string, r io.Reader, reqTime time.Duration, reqErr error, pollID uint64, pollFinished chan<- uint64) {
log.Debugf("poll %v %v handle start\n", pollID, time.Now())
[16/20] incubator-trafficcontrol git commit: Remove TM2 wrong comments
Posted by ne...@apache.org.
Remove TM2 wrong comments
Project: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/commit/b2cb7368
Tree: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/tree/b2cb7368
Diff: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/diff/b2cb7368
Branch: refs/heads/master
Commit: b2cb73685c83c30d830b9d2fbc25ea51cd827c5b
Parents: 34863ed
Author: Robert Butts <ro...@gmail.com>
Authored: Mon Jan 23 09:58:23 2017 -0700
Committer: Dave Neuman <ne...@apache.org>
Committed: Wed Jan 25 10:29:46 2017 -0700
----------------------------------------------------------------------
.../experimental/traffic_monitor/deliveryservice/stat.go | 1 -
.../experimental/traffic_monitor/threadsafe/resultstathistory.go | 2 --
2 files changed, 3 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/b2cb7368/traffic_monitor/experimental/traffic_monitor/deliveryservice/stat.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/deliveryservice/stat.go b/traffic_monitor/experimental/traffic_monitor/deliveryservice/stat.go
index 30e1f74..1958f05 100644
--- a/traffic_monitor/experimental/traffic_monitor/deliveryservice/stat.go
+++ b/traffic_monitor/experimental/traffic_monitor/deliveryservice/stat.go
@@ -39,7 +39,6 @@ import (
// TODO remove 'ds' and 'stat' from names
// Stats is the JSON-serialisable representation of delivery service Stats. It maps delivery service names to individual stat objects.
-// TODO remove DeliveryService and set type to the map directly, or add other members
type Stats struct {
DeliveryService map[enum.DeliveryServiceName]dsdata.Stat `json:"deliveryService"`
Time time.Time `json:"-"`
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/b2cb7368/traffic_monitor/experimental/traffic_monitor/threadsafe/resultstathistory.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/threadsafe/resultstathistory.go b/traffic_monitor/experimental/traffic_monitor/threadsafe/resultstathistory.go
index 966b3ac..cf8f996 100644
--- a/traffic_monitor/experimental/traffic_monitor/threadsafe/resultstathistory.go
+++ b/traffic_monitor/experimental/traffic_monitor/threadsafe/resultstathistory.go
@@ -28,7 +28,6 @@ import (
// ResultStatHistory provides safe access for multiple goroutines readers and a single writer to a stored HistoryHistory object.
// This could be made lock-free, if the performance was necessary
-// TODO add separate locks for Caches and Deliveryservice maps?
type ResultStatHistory struct {
history *cache.ResultStatHistory
m *sync.RWMutex
@@ -56,7 +55,6 @@ func (h *ResultStatHistory) Set(v cache.ResultStatHistory) {
// ResultStatHistory provides safe access for multiple goroutines readers and a single writer to a stored HistoryHistory object.
// This could be made lock-free, if the performance was necessary
-// TODO add separate locks for Caches and Deliveryservice maps?
type ResultInfoHistory struct {
history *cache.ResultInfoHistory
m *sync.RWMutex
[13/20] incubator-trafficcontrol git commit: Remove TM2 unused funcs
Posted by ne...@apache.org.
Remove TM2 unused funcs
Project: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/commit/4fec5996
Tree: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/tree/4fec5996
Diff: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/diff/4fec5996
Branch: refs/heads/master
Commit: 4fec5996b894040498cd4df1df091a7093cfd6a2
Parents: 1ffb301
Author: Robert Butts <ro...@gmail.com>
Authored: Mon Jan 23 15:22:15 2017 -0700
Committer: Dave Neuman <ne...@apache.org>
Committed: Wed Jan 25 10:29:46 2017 -0700
----------------------------------------------------------------------
.../experimental/traffic_monitor/cache/data.go | 8 ------
.../traffic_monitor/health/cache_health.go | 30 --------------------
.../traffic_monitor/manager/datarequest.go | 12 --------
3 files changed, 50 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/4fec5996/traffic_monitor/experimental/traffic_monitor/cache/data.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/cache/data.go b/traffic_monitor/experimental/traffic_monitor/cache/data.go
index bad75a0..81b938a 100644
--- a/traffic_monitor/experimental/traffic_monitor/cache/data.go
+++ b/traffic_monitor/experimental/traffic_monitor/cache/data.go
@@ -178,14 +178,6 @@ func ToInfo(r Result) ResultInfo {
}
}
-func toInfos(rs []Result) []ResultInfo {
- infos := make([]ResultInfo, len(rs), len(rs))
- for i, r := range rs {
- infos[i] = ToInfo(r)
- }
- return infos
-}
-
func copyResultInfos(a []ResultInfo) []ResultInfo {
b := make([]ResultInfo, len(a), len(a))
copy(b, a)
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/4fec5996/traffic_monitor/experimental/traffic_monitor/health/cache_health.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/health/cache_health.go b/traffic_monitor/experimental/traffic_monitor/health/cache_health.go
index 15d5b9a..0a00838 100644
--- a/traffic_monitor/experimental/traffic_monitor/health/cache_health.go
+++ b/traffic_monitor/experimental/traffic_monitor/health/cache_health.go
@@ -23,7 +23,6 @@ import (
"fmt"
"strconv"
"strings"
- "time"
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/common/log"
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/cache"
@@ -95,35 +94,6 @@ func GetVitals(newResult *cache.Result, prevResult *cache.Result, mc *to.Traffic
// log.Infoln(newResult.Id, "BytesOut", newResult.Vitals.BytesOut, "BytesIn", newResult.Vitals.BytesIn, "Kbps", newResult.Vitals.KbpsOut, "max", newResult.Vitals.MaxKbpsOut)
}
-// getKbpsThreshold returns the numeric kbps threshold, from the Traffic Ops string value. If there is a parse error, it logs a warning and returns the max floating point number, signifying no limit
-// TODO add float64 to Traffic Ops Client interface
-func getKbpsThreshold(threshStr string) (int64, bool) {
- if len(threshStr) == 0 {
- return 0, false
- }
- if threshStr[0] == '>' {
- threshStr = threshStr[1:]
- }
- thresh, err := strconv.ParseInt(threshStr, 10, 64)
- if err != nil {
- return 0, false
- }
- return thresh, true
-}
-
-// TODO add time.Duration to Traffic Ops Client interface
-func getQueryThreshold(threshInt int64) (time.Duration, bool) {
- if threshInt == 0 {
- return 0, false
- }
- return time.Duration(threshInt) * time.Millisecond, true
-}
-
-func cacheCapacityKbps(result cache.Result) int64 {
- kbpsInMbps := int64(1000)
- return int64(result.Astats.System.InfSpeed) * kbpsInMbps
-}
-
// EvalCache returns whether the given cache should be marked available, a string describing why, and which stat exceeded a threshold. The `stats` may be nil, for pollers which don't poll stats.
// The availability of EvalCache MAY NOT be used to directly set the cache's local availability, because the threshold stats may not be part of the poller which produced the result. Rather, if the cache was previously unavailable from a threshold, it must be verified that threshold stat is in the results before setting the cache to available.
// TODO change to return a `cache.AvailableStatus`
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/4fec5996/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go b/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
index 3cc4c4a..30fcc47 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
@@ -557,18 +557,6 @@ func srvTRConfig(opsConfig OpsConfigThreadsafe, toSession towrap.ITrafficOpsSess
return toSession.CRConfigRaw(cdnName)
}
-func makeWrapAll(errorCount threadsafe.Uint, unpolledCaches threadsafe.UnpolledCaches) func(http.HandlerFunc) http.HandlerFunc {
- return func(f http.HandlerFunc) http.HandlerFunc {
- return wrapUnpolledCheck(unpolledCaches, errorCount, f)
- }
-}
-
-func makeCrConfigHandler(wrapper func(http.HandlerFunc) http.HandlerFunc, errorCount threadsafe.Uint, opsConfig OpsConfigThreadsafe, toSession towrap.ITrafficOpsSession) http.HandlerFunc {
- return wrapper(WrapErr(errorCount, func() ([]byte, error) {
- return srvTRConfig(opsConfig, toSession)
- }, ContentTypeJSON))
-}
-
func srvTRState(params url.Values, localStates peer.CRStatesThreadsafe, combinedStates peer.CRStatesThreadsafe) ([]byte, error) {
if _, raw := params["raw"]; raw {
return srvTRStateSelf(localStates)
[18/20] incubator-trafficcontrol git commit: Rename TM2
manager/healthresult.go to health.go
Posted by ne...@apache.org.
Rename TM2 manager/healthresult.go to health.go
Project: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/commit/b575ec15
Tree: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/tree/b575ec15
Diff: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/diff/b575ec15
Branch: refs/heads/master
Commit: b575ec15e4eda2dc76810657b5a3dec75d853933
Parents: cc10a02
Author: Robert Butts <ro...@gmail.com>
Authored: Tue Jan 24 10:45:20 2017 -0700
Committer: Dave Neuman <ne...@apache.org>
Committed: Wed Jan 25 10:29:46 2017 -0700
----------------------------------------------------------------------
.../traffic_monitor/manager/health.go | 267 +++++++++++++++++++
.../traffic_monitor/manager/healthresult.go | 267 -------------------
2 files changed, 267 insertions(+), 267 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/b575ec15/traffic_monitor/experimental/traffic_monitor/manager/health.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/health.go b/traffic_monitor/experimental/traffic_monitor/manager/health.go
new file mode 100644
index 0000000..63248e8
--- /dev/null
+++ b/traffic_monitor/experimental/traffic_monitor/manager/health.go
@@ -0,0 +1,267 @@
+package manager
+
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+import (
+ "sync"
+ "time"
+
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/common/log"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/cache"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/config"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/enum"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/health"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/peer"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/threadsafe"
+ todata "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/trafficopsdata"
+)
+
+// DurationMap represents a map of cache names to durations
+type DurationMap map[enum.CacheName]time.Duration
+
+// DurationMapThreadsafe wraps a DurationMap in an object safe for a single writer and multiple readers
+type DurationMapThreadsafe struct {
+ durationMap *DurationMap
+ m *sync.RWMutex
+}
+
+// Copy copies this duration map.
+func (a DurationMap) Copy() DurationMap {
+ b := DurationMap{}
+ for k, v := range a {
+ b[k] = v
+ }
+ return b
+}
+
+// NewDurationMapThreadsafe returns a new DurationMapThreadsafe safe for multiple readers and a single writer goroutine.
+func NewDurationMapThreadsafe() DurationMapThreadsafe {
+ m := DurationMap{}
+ return DurationMapThreadsafe{m: &sync.RWMutex{}, durationMap: &m}
+}
+
+// Get returns the duration map. Callers MUST NOT mutate. If mutation is necessary, call DurationMap.Copy().
+func (o *DurationMapThreadsafe) Get() DurationMap {
+ o.m.RLock()
+ defer o.m.RUnlock()
+ return *o.durationMap
+}
+
+// Set sets the internal duration map. This MUST NOT be called by multiple goroutines.
+func (o *DurationMapThreadsafe) Set(d DurationMap) {
+ o.m.Lock()
+ *o.durationMap = d
+ o.m.Unlock()
+}
+
+// StartHealthResultManager starts the goroutine which listens for health results.
+// Note this polls the brief stat endpoint from ATS Astats, not the full stats.
+// This poll should be quicker and less computationally expensive for ATS, but
+// doesn't include all stat data needed for e.g. delivery service calculations.4
+// Returns the last health durations, events, the local cache statuses, and the health result history.
+func StartHealthResultManager(
+ cacheHealthChan <-chan cache.Result,
+ toData todata.TODataThreadsafe,
+ localStates peer.CRStatesThreadsafe,
+ monitorConfig TrafficMonitorConfigMapThreadsafe,
+ peerStates peer.CRStatesPeersThreadsafe,
+ combinedStates peer.CRStatesThreadsafe,
+ fetchCount threadsafe.Uint,
+ errorCount threadsafe.Uint,
+ cfg config.Config,
+ events threadsafe.Events,
+ localCacheStatus threadsafe.CacheAvailableStatus,
+) (DurationMapThreadsafe, threadsafe.ResultHistory) {
+ lastHealthDurations := NewDurationMapThreadsafe()
+ healthHistory := threadsafe.NewResultHistory()
+ go healthResultManagerListen(
+ cacheHealthChan,
+ toData,
+ localStates,
+ lastHealthDurations,
+ healthHistory,
+ monitorConfig,
+ peerStates,
+ combinedStates,
+ fetchCount,
+ errorCount,
+ events,
+ localCacheStatus,
+ cfg,
+ )
+ return lastHealthDurations, healthHistory
+}
+
+func healthResultManagerListen(
+ cacheHealthChan <-chan cache.Result,
+ toData todata.TODataThreadsafe,
+ localStates peer.CRStatesThreadsafe,
+ lastHealthDurations DurationMapThreadsafe,
+ healthHistory threadsafe.ResultHistory,
+ monitorConfig TrafficMonitorConfigMapThreadsafe,
+ peerStates peer.CRStatesPeersThreadsafe,
+ combinedStates peer.CRStatesThreadsafe,
+ fetchCount threadsafe.Uint,
+ errorCount threadsafe.Uint,
+ events threadsafe.Events,
+ localCacheStatus threadsafe.CacheAvailableStatus,
+ cfg config.Config,
+) {
+ lastHealthEndTimes := map[enum.CacheName]time.Time{}
+ // This reads at least 1 value from the cacheHealthChan. Then, we loop, and try to read from the channel some more. If there's nothing to read, we hit `default` and process. If there is stuff to read, we read it, then inner-loop trying to read more. If we're continuously reading and the channel is never empty, and we hit the tick time, process anyway even though the channel isn't empty, to prevent never processing (starvation).
+ var ticker *time.Ticker
+
+ process := func(results []cache.Result) {
+ processHealthResult(
+ cacheHealthChan,
+ toData,
+ localStates,
+ lastHealthDurations,
+ monitorConfig,
+ peerStates,
+ combinedStates,
+ fetchCount,
+ errorCount,
+ events,
+ localCacheStatus,
+ lastHealthEndTimes,
+ healthHistory,
+ results,
+ cfg,
+ )
+ }
+
+ for {
+ var results []cache.Result
+ results = append(results, <-cacheHealthChan)
+ if ticker != nil {
+ ticker.Stop()
+ }
+ ticker = time.NewTicker(cfg.HealthFlushInterval)
+ innerLoop:
+ for {
+ select {
+ case <-ticker.C:
+ log.Warnf("Health Result Manager flushing queued results\n")
+ process(results)
+ break innerLoop
+ default:
+ select {
+ case r := <-cacheHealthChan:
+ results = append(results, r)
+ default:
+ process(results)
+ break innerLoop
+ }
+ }
+ }
+ }
+}
+
+// processHealthResult processes the given health results, adding their stats to the CacheAvailableStatus. Note this is NOT threadsafe, because it non-atomically gets CacheAvailableStatuses, Events, LastHealthDurations and later updates them. This MUST NOT be called from multiple threads.
+func processHealthResult(
+ cacheHealthChan <-chan cache.Result,
+ toData todata.TODataThreadsafe,
+ localStates peer.CRStatesThreadsafe,
+ lastHealthDurationsThreadsafe DurationMapThreadsafe,
+ monitorConfig TrafficMonitorConfigMapThreadsafe,
+ peerStates peer.CRStatesPeersThreadsafe,
+ combinedStates peer.CRStatesThreadsafe,
+ fetchCount threadsafe.Uint,
+ errorCount threadsafe.Uint,
+ events threadsafe.Events,
+ localCacheStatusThreadsafe threadsafe.CacheAvailableStatus,
+ lastHealthEndTimes map[enum.CacheName]time.Time,
+ healthHistory threadsafe.ResultHistory,
+ results []cache.Result,
+ cfg config.Config,
+) {
+ if len(results) == 0 {
+ return
+ }
+ defer func() {
+ for _, r := range results {
+ log.Debugf("poll %v %v finish\n", r.PollID, time.Now())
+ r.PollFinished <- r.PollID
+ }
+ }()
+
+ toDataCopy := toData.Get() // create a copy, so the same data used for all processing of this cache health result
+ monitorConfigCopy := monitorConfig.Get()
+ healthHistoryCopy := healthHistory.Get().Copy()
+ for i, healthResult := range results {
+ log.Debugf("poll %v %v healthresultman start\n", healthResult.PollID, time.Now())
+ fetchCount.Inc()
+ var prevResult cache.Result
+ healthResultHistory := healthHistoryCopy[healthResult.ID]
+ if len(healthResultHistory) != 0 {
+ prevResult = healthResultHistory[len(healthResultHistory)-1]
+ }
+
+ if healthResult.Error == nil {
+ health.GetVitals(&healthResult, &prevResult, &monitorConfigCopy)
+ results[i] = healthResult
+ }
+
+ maxHistory := uint64(monitorConfigCopy.Profile[monitorConfigCopy.TrafficServer[string(healthResult.ID)].Profile].Parameters.HistoryCount)
+ if maxHistory < 1 {
+ log.Warnf("processHealthResult got history count %v for %v, setting to 1\n", maxHistory, healthResult.ID)
+ maxHistory = 1
+ }
+
+ healthHistoryCopy[healthResult.ID] = pruneHistory(append([]cache.Result{healthResult}, healthHistoryCopy[healthResult.ID]...), maxHistory)
+ }
+
+ calcAvailability(results, "health", nil, monitorConfigCopy, toDataCopy, localCacheStatusThreadsafe, localStates, events)
+
+ healthHistory.Set(healthHistoryCopy)
+ // TODO determine if we should combineCrStates() here
+
+ lastHealthDurations := lastHealthDurationsThreadsafe.Get().Copy()
+ for _, healthResult := range results {
+ if lastHealthStart, ok := lastHealthEndTimes[healthResult.ID]; ok {
+ d := time.Since(lastHealthStart)
+ lastHealthDurations[healthResult.ID] = d
+ }
+ lastHealthEndTimes[healthResult.ID] = time.Now()
+ }
+ lastHealthDurationsThreadsafe.Set(lastHealthDurations)
+}
+
+// calculateDeliveryServiceState calculates the state of delivery services from the new cache state data `cacheState` and the CRConfig data `deliveryServiceServers` and puts the calculated state in the outparam `deliveryServiceStates`
+func CalculateDeliveryServiceState(deliveryServiceServers map[enum.DeliveryServiceName][]enum.CacheName, states peer.CRStatesThreadsafe) {
+ deliveryServices := states.GetDeliveryServices()
+ for deliveryServiceName, deliveryServiceState := range deliveryServices {
+ if _, ok := deliveryServiceServers[deliveryServiceName]; !ok {
+ // log.Errorf("CRConfig does not have delivery service %s, but traffic monitor poller does; skipping\n", deliveryServiceName)
+ continue
+ }
+ deliveryServiceState.IsAvailable = false
+ deliveryServiceState.DisabledLocations = []enum.CacheName{} // it's important this isn't nil, so it serialises to the JSON `[]` instead of `null`
+ for _, server := range deliveryServiceServers[deliveryServiceName] {
+ if available, _ := states.GetCache(server); available.IsAvailable {
+ deliveryServiceState.IsAvailable = true
+ } else {
+ deliveryServiceState.DisabledLocations = append(deliveryServiceState.DisabledLocations, server)
+ }
+ }
+ states.SetDeliveryService(deliveryServiceName, deliveryServiceState)
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/b575ec15/traffic_monitor/experimental/traffic_monitor/manager/healthresult.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/healthresult.go b/traffic_monitor/experimental/traffic_monitor/manager/healthresult.go
deleted file mode 100644
index 63248e8..0000000
--- a/traffic_monitor/experimental/traffic_monitor/manager/healthresult.go
+++ /dev/null
@@ -1,267 +0,0 @@
-package manager
-
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing,
- * software distributed under the License is distributed on an
- * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
- * KIND, either express or implied. See the License for the
- * specific language governing permissions and limitations
- * under the License.
- */
-
-import (
- "sync"
- "time"
-
- "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/common/log"
- "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/cache"
- "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/config"
- "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/enum"
- "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/health"
- "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/peer"
- "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/threadsafe"
- todata "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/trafficopsdata"
-)
-
-// DurationMap represents a map of cache names to durations
-type DurationMap map[enum.CacheName]time.Duration
-
-// DurationMapThreadsafe wraps a DurationMap in an object safe for a single writer and multiple readers
-type DurationMapThreadsafe struct {
- durationMap *DurationMap
- m *sync.RWMutex
-}
-
-// Copy copies this duration map.
-func (a DurationMap) Copy() DurationMap {
- b := DurationMap{}
- for k, v := range a {
- b[k] = v
- }
- return b
-}
-
-// NewDurationMapThreadsafe returns a new DurationMapThreadsafe safe for multiple readers and a single writer goroutine.
-func NewDurationMapThreadsafe() DurationMapThreadsafe {
- m := DurationMap{}
- return DurationMapThreadsafe{m: &sync.RWMutex{}, durationMap: &m}
-}
-
-// Get returns the duration map. Callers MUST NOT mutate. If mutation is necessary, call DurationMap.Copy().
-func (o *DurationMapThreadsafe) Get() DurationMap {
- o.m.RLock()
- defer o.m.RUnlock()
- return *o.durationMap
-}
-
-// Set sets the internal duration map. This MUST NOT be called by multiple goroutines.
-func (o *DurationMapThreadsafe) Set(d DurationMap) {
- o.m.Lock()
- *o.durationMap = d
- o.m.Unlock()
-}
-
-// StartHealthResultManager starts the goroutine which listens for health results.
-// Note this polls the brief stat endpoint from ATS Astats, not the full stats.
-// This poll should be quicker and less computationally expensive for ATS, but
-// doesn't include all stat data needed for e.g. delivery service calculations.4
-// Returns the last health durations, events, the local cache statuses, and the health result history.
-func StartHealthResultManager(
- cacheHealthChan <-chan cache.Result,
- toData todata.TODataThreadsafe,
- localStates peer.CRStatesThreadsafe,
- monitorConfig TrafficMonitorConfigMapThreadsafe,
- peerStates peer.CRStatesPeersThreadsafe,
- combinedStates peer.CRStatesThreadsafe,
- fetchCount threadsafe.Uint,
- errorCount threadsafe.Uint,
- cfg config.Config,
- events threadsafe.Events,
- localCacheStatus threadsafe.CacheAvailableStatus,
-) (DurationMapThreadsafe, threadsafe.ResultHistory) {
- lastHealthDurations := NewDurationMapThreadsafe()
- healthHistory := threadsafe.NewResultHistory()
- go healthResultManagerListen(
- cacheHealthChan,
- toData,
- localStates,
- lastHealthDurations,
- healthHistory,
- monitorConfig,
- peerStates,
- combinedStates,
- fetchCount,
- errorCount,
- events,
- localCacheStatus,
- cfg,
- )
- return lastHealthDurations, healthHistory
-}
-
-func healthResultManagerListen(
- cacheHealthChan <-chan cache.Result,
- toData todata.TODataThreadsafe,
- localStates peer.CRStatesThreadsafe,
- lastHealthDurations DurationMapThreadsafe,
- healthHistory threadsafe.ResultHistory,
- monitorConfig TrafficMonitorConfigMapThreadsafe,
- peerStates peer.CRStatesPeersThreadsafe,
- combinedStates peer.CRStatesThreadsafe,
- fetchCount threadsafe.Uint,
- errorCount threadsafe.Uint,
- events threadsafe.Events,
- localCacheStatus threadsafe.CacheAvailableStatus,
- cfg config.Config,
-) {
- lastHealthEndTimes := map[enum.CacheName]time.Time{}
- // This reads at least 1 value from the cacheHealthChan. Then, we loop, and try to read from the channel some more. If there's nothing to read, we hit `default` and process. If there is stuff to read, we read it, then inner-loop trying to read more. If we're continuously reading and the channel is never empty, and we hit the tick time, process anyway even though the channel isn't empty, to prevent never processing (starvation).
- var ticker *time.Ticker
-
- process := func(results []cache.Result) {
- processHealthResult(
- cacheHealthChan,
- toData,
- localStates,
- lastHealthDurations,
- monitorConfig,
- peerStates,
- combinedStates,
- fetchCount,
- errorCount,
- events,
- localCacheStatus,
- lastHealthEndTimes,
- healthHistory,
- results,
- cfg,
- )
- }
-
- for {
- var results []cache.Result
- results = append(results, <-cacheHealthChan)
- if ticker != nil {
- ticker.Stop()
- }
- ticker = time.NewTicker(cfg.HealthFlushInterval)
- innerLoop:
- for {
- select {
- case <-ticker.C:
- log.Warnf("Health Result Manager flushing queued results\n")
- process(results)
- break innerLoop
- default:
- select {
- case r := <-cacheHealthChan:
- results = append(results, r)
- default:
- process(results)
- break innerLoop
- }
- }
- }
- }
-}
-
-// processHealthResult processes the given health results, adding their stats to the CacheAvailableStatus. Note this is NOT threadsafe, because it non-atomically gets CacheAvailableStatuses, Events, LastHealthDurations and later updates them. This MUST NOT be called from multiple threads.
-func processHealthResult(
- cacheHealthChan <-chan cache.Result,
- toData todata.TODataThreadsafe,
- localStates peer.CRStatesThreadsafe,
- lastHealthDurationsThreadsafe DurationMapThreadsafe,
- monitorConfig TrafficMonitorConfigMapThreadsafe,
- peerStates peer.CRStatesPeersThreadsafe,
- combinedStates peer.CRStatesThreadsafe,
- fetchCount threadsafe.Uint,
- errorCount threadsafe.Uint,
- events threadsafe.Events,
- localCacheStatusThreadsafe threadsafe.CacheAvailableStatus,
- lastHealthEndTimes map[enum.CacheName]time.Time,
- healthHistory threadsafe.ResultHistory,
- results []cache.Result,
- cfg config.Config,
-) {
- if len(results) == 0 {
- return
- }
- defer func() {
- for _, r := range results {
- log.Debugf("poll %v %v finish\n", r.PollID, time.Now())
- r.PollFinished <- r.PollID
- }
- }()
-
- toDataCopy := toData.Get() // create a copy, so the same data used for all processing of this cache health result
- monitorConfigCopy := monitorConfig.Get()
- healthHistoryCopy := healthHistory.Get().Copy()
- for i, healthResult := range results {
- log.Debugf("poll %v %v healthresultman start\n", healthResult.PollID, time.Now())
- fetchCount.Inc()
- var prevResult cache.Result
- healthResultHistory := healthHistoryCopy[healthResult.ID]
- if len(healthResultHistory) != 0 {
- prevResult = healthResultHistory[len(healthResultHistory)-1]
- }
-
- if healthResult.Error == nil {
- health.GetVitals(&healthResult, &prevResult, &monitorConfigCopy)
- results[i] = healthResult
- }
-
- maxHistory := uint64(monitorConfigCopy.Profile[monitorConfigCopy.TrafficServer[string(healthResult.ID)].Profile].Parameters.HistoryCount)
- if maxHistory < 1 {
- log.Warnf("processHealthResult got history count %v for %v, setting to 1\n", maxHistory, healthResult.ID)
- maxHistory = 1
- }
-
- healthHistoryCopy[healthResult.ID] = pruneHistory(append([]cache.Result{healthResult}, healthHistoryCopy[healthResult.ID]...), maxHistory)
- }
-
- calcAvailability(results, "health", nil, monitorConfigCopy, toDataCopy, localCacheStatusThreadsafe, localStates, events)
-
- healthHistory.Set(healthHistoryCopy)
- // TODO determine if we should combineCrStates() here
-
- lastHealthDurations := lastHealthDurationsThreadsafe.Get().Copy()
- for _, healthResult := range results {
- if lastHealthStart, ok := lastHealthEndTimes[healthResult.ID]; ok {
- d := time.Since(lastHealthStart)
- lastHealthDurations[healthResult.ID] = d
- }
- lastHealthEndTimes[healthResult.ID] = time.Now()
- }
- lastHealthDurationsThreadsafe.Set(lastHealthDurations)
-}
-
-// calculateDeliveryServiceState calculates the state of delivery services from the new cache state data `cacheState` and the CRConfig data `deliveryServiceServers` and puts the calculated state in the outparam `deliveryServiceStates`
-func CalculateDeliveryServiceState(deliveryServiceServers map[enum.DeliveryServiceName][]enum.CacheName, states peer.CRStatesThreadsafe) {
- deliveryServices := states.GetDeliveryServices()
- for deliveryServiceName, deliveryServiceState := range deliveryServices {
- if _, ok := deliveryServiceServers[deliveryServiceName]; !ok {
- // log.Errorf("CRConfig does not have delivery service %s, but traffic monitor poller does; skipping\n", deliveryServiceName)
- continue
- }
- deliveryServiceState.IsAvailable = false
- deliveryServiceState.DisabledLocations = []enum.CacheName{} // it's important this isn't nil, so it serialises to the JSON `[]` instead of `null`
- for _, server := range deliveryServiceServers[deliveryServiceName] {
- if available, _ := states.GetCache(server); available.IsAvailable {
- deliveryServiceState.IsAvailable = true
- } else {
- deliveryServiceState.DisabledLocations = append(deliveryServiceState.DisabledLocations, server)
- }
- }
- states.SetDeliveryService(deliveryServiceName, deliveryServiceState)
- }
-}
[06/20] incubator-trafficcontrol git commit: Add TM2 dynamic param
health thresholds
Posted by ne...@apache.org.
Add TM2 dynamic param health thresholds
Project: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/commit/8a7bed40
Tree: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/tree/8a7bed40
Diff: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/diff/8a7bed40
Branch: refs/heads/master
Commit: 8a7bed40c856dbafe441e599b4ca335d4247e1a2
Parents: 9c6dbea
Author: Robert Butts <ro...@gmail.com>
Authored: Fri Jan 20 10:00:52 2017 -0700
Committer: Dave Neuman <ne...@apache.org>
Committed: Wed Jan 25 10:29:46 2017 -0700
----------------------------------------------------------------------
.../experimental/traffic_monitor/cache/cache.go | 198 +++++++++----------
.../experimental/traffic_monitor/cache/data.go | 8 +-
.../experimental/traffic_monitor/enum/enum.go | 35 ++++
.../traffic_monitor/health/cache_health.go | 123 +++++++++---
.../traffic_monitor/manager/datarequest.go | 41 +---
.../traffic_monitor/manager/healthresult.go | 112 ++++++-----
.../traffic_monitor/manager/manager.go | 11 +-
.../traffic_monitor/manager/peer.go | 6 +-
.../traffic_monitor/manager/stathistory.go | 62 +++++-
.../traffic_monitor/threadsafe/events.go | 3 +-
traffic_ops/client/traffic_monitor_config.go | 96 +++++++--
11 files changed, 452 insertions(+), 243 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/8a7bed40/traffic_monitor/experimental/traffic_monitor/cache/cache.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/cache/cache.go b/traffic_monitor/experimental/traffic_monitor/cache/cache.go
index 0aea799..541edaa 100644
--- a/traffic_monitor/experimental/traffic_monitor/cache/cache.go
+++ b/traffic_monitor/experimental/traffic_monitor/cache/cache.go
@@ -117,6 +117,84 @@ type Filter interface {
const nsPerMs = 1000000
+type StatComputeFunc func(resultInfo ResultInfo, serverInfo to.TrafficServer, serverProfile to.TMProfile, combinedState peer.IsAvailable) interface{}
+
+// ComputedStats returns a map of cache stats which are computed by Traffic Monitor (rather than returned literally from ATS), mapped to the func to compute them.
+func ComputedStats() map[string]StatComputeFunc {
+ return map[string]StatComputeFunc{
+ "availableBandwidthInKbps": func(info ResultInfo, serverInfo to.TrafficServer, serverProfile to.TMProfile, combinedState peer.IsAvailable) interface{} {
+ return info.Vitals.MaxKbpsOut - info.Vitals.KbpsOut
+ },
+
+ "availableBandwidthInMbps": func(info ResultInfo, serverInfo to.TrafficServer, serverProfile to.TMProfile, combinedState peer.IsAvailable) interface{} {
+ return (info.Vitals.MaxKbpsOut - info.Vitals.KbpsOut) / 1000
+ },
+ "bandwidth": func(info ResultInfo, serverInfo to.TrafficServer, serverProfile to.TMProfile, combinedState peer.IsAvailable) interface{} {
+ return info.Vitals.KbpsOut
+ },
+ "error-string": func(info ResultInfo, serverInfo to.TrafficServer, serverProfile to.TMProfile, combinedState peer.IsAvailable) interface{} {
+ if info.Error != nil {
+ return info.Error.Error()
+ }
+ return "false"
+ },
+ "isAvailable": func(info ResultInfo, serverInfo to.TrafficServer, serverProfile to.TMProfile, combinedState peer.IsAvailable) interface{} {
+ return combinedState.IsAvailable // if the cache is missing, default to false
+ },
+ "isHealthy": func(info ResultInfo, serverInfo to.TrafficServer, serverProfile to.TMProfile, combinedState peer.IsAvailable) interface{} {
+ if enum.CacheStatusFromString(serverInfo.Status) == enum.CacheStatusAdminDown {
+ return true
+ }
+ return combinedState.IsAvailable
+ },
+ "kbps": func(info ResultInfo, serverInfo to.TrafficServer, serverProfile to.TMProfile, combinedState peer.IsAvailable) interface{} {
+ return info.Vitals.KbpsOut
+ },
+ "loadavg": func(info ResultInfo, serverInfo to.TrafficServer, serverProfile to.TMProfile, combinedState peer.IsAvailable) interface{} {
+ return info.Vitals.LoadAvg
+ },
+ "maxKbps": func(info ResultInfo, serverInfo to.TrafficServer, serverProfile to.TMProfile, combinedState peer.IsAvailable) interface{} {
+ return info.Vitals.MaxKbpsOut
+ },
+ "queryTime": func(info ResultInfo, serverInfo to.TrafficServer, serverProfile to.TMProfile, combinedState peer.IsAvailable) interface{} {
+ return info.RequestTime.Nanoseconds() / nsPerMs
+ },
+ "stateUrl": func(info ResultInfo, serverInfo to.TrafficServer, serverProfile to.TMProfile, combinedState peer.IsAvailable) interface{} {
+ return serverProfile.Parameters.HealthPollingURL
+ },
+ "status": func(info ResultInfo, serverInfo to.TrafficServer, serverProfile to.TMProfile, combinedState peer.IsAvailable) interface{} {
+ return serverInfo.Status
+ },
+ "system.astatsLoad": func(info ResultInfo, serverInfo to.TrafficServer, serverProfile to.TMProfile, combinedState peer.IsAvailable) interface{} {
+ return info.System.AstatsLoad
+ },
+ "system.configReloadRequests": func(info ResultInfo, serverInfo to.TrafficServer, serverProfile to.TMProfile, combinedState peer.IsAvailable) interface{} {
+ return info.System.ConfigLoadRequest
+ },
+ "system.configReloads": func(info ResultInfo, serverInfo to.TrafficServer, serverProfile to.TMProfile, combinedState peer.IsAvailable) interface{} {
+ return info.System.ConfigReloads
+ },
+ "system.inf.name": func(info ResultInfo, serverInfo to.TrafficServer, serverProfile to.TMProfile, combinedState peer.IsAvailable) interface{} {
+ return info.System.InfName
+ },
+ "system.inf.speed": func(info ResultInfo, serverInfo to.TrafficServer, serverProfile to.TMProfile, combinedState peer.IsAvailable) interface{} {
+ return info.System.InfSpeed
+ },
+ "system.lastReload": func(info ResultInfo, serverInfo to.TrafficServer, serverProfile to.TMProfile, combinedState peer.IsAvailable) interface{} {
+ return info.System.LastReload
+ },
+ "system.lastReloadRequest": func(info ResultInfo, serverInfo to.TrafficServer, serverProfile to.TMProfile, combinedState peer.IsAvailable) interface{} {
+ return info.System.LastReloadRequest
+ },
+ "system.proc.loadavg": func(info ResultInfo, serverInfo to.TrafficServer, serverProfile to.TMProfile, combinedState peer.IsAvailable) interface{} {
+ return info.System.ProcLoadavg
+ },
+ "system.proc.net.dev": func(info ResultInfo, serverInfo to.TrafficServer, serverProfile to.TMProfile, combinedState peer.IsAvailable) interface{} {
+ return info.System.ProcNetDev
+ },
+ }
+}
+
// StatsMarshall encodes the stats in JSON, encoding up to historyCount of each stat. If statsToUse is empty, all stats are encoded; otherwise, only the given stats are encoded. If wildcard is true, stats which contain the text in each statsToUse are returned, instead of exact stat names. If cacheType is not CacheTypeInvalid, only stats for the given type are returned. If hosts is not empty, only the given hosts are returned.
func StatsMarshall(statResultHistory ResultStatHistory, statInfo ResultInfoHistory, combinedStates peer.Crstates, monitorConfig to.TrafficMonitorConfigMap, statMaxKbpses Kbpses, filter Filter, params url.Values) ([]byte, error) {
stats := Stats{
@@ -124,6 +202,8 @@ func StatsMarshall(statResultHistory ResultStatHistory, statInfo ResultInfoHisto
Caches: map[enum.CacheName]map[string][]ResultStatVal{},
}
+ computedStats := ComputedStats()
+
// TODO in 1.0, stats are divided into 'location', 'cache', and 'type'. 'cache' are hidden by default.
for id, history := range statResultHistory {
@@ -153,7 +233,18 @@ func StatsMarshall(statResultHistory ResultStatHistory, statInfo ResultInfoHisto
if !filter.UseCache(id) {
continue
}
- for i, info := range infos {
+
+ serverInfo, ok := monitorConfig.TrafficServer[string(id)]
+ if !ok {
+ log.Warnf("cache.StatsMarshall server %s missing from monitorConfig\n", id)
+ }
+
+ serverProfile, ok := monitorConfig.Profile[serverInfo.Profile]
+ if !ok {
+ log.Warnf("cache.StatsMarshall server %s missing profile in monitorConfig\n", id)
+ }
+
+ for i, resultInfo := range infos {
if !filter.WithinStatHistoryMax(i + 1) {
break
}
@@ -161,108 +252,13 @@ func StatsMarshall(statResultHistory ResultStatHistory, statInfo ResultInfoHisto
stats.Caches[id] = map[string][]ResultStatVal{}
}
- t := info.Time
+ t := resultInfo.Time
- if stat := "availableBandwidthInKbps"; filter.UseStat(stat) {
- v := info.Vitals.MaxKbpsOut - info.Vitals.KbpsOut
- stats.Caches[id][stat] = append(stats.Caches[id][stat], ResultStatVal{Val: v, Time: t, Span: 1})
- }
- if stat := "availableBandwidthInMbps"; filter.UseStat(stat) {
- v := (info.Vitals.MaxKbpsOut - info.Vitals.KbpsOut) / 1000
- stats.Caches[id][stat] = append(stats.Caches[id][stat], ResultStatVal{Val: v, Time: t, Span: 1})
- }
- if stat := "bandwidth"; filter.UseStat(stat) {
- v := info.Vitals
- stats.Caches[id][stat] = append(stats.Caches[id][stat], ResultStatVal{Val: v, Time: t, Span: 1})
- }
- if stat := "error-string"; filter.UseStat(stat) {
- v := ""
- if info.Error != nil {
- v = info.Error.Error()
- } else {
- v = "false"
- }
- stats.Caches[id][stat] = append(stats.Caches[id][stat], ResultStatVal{Val: v, Time: t, Span: 1})
- }
- if stat := "isAvailable"; filter.UseStat(stat) {
- v := combinedStates.Caches[id].IsAvailable // if the cache is missing, default to false
- stats.Caches[id][stat] = append(stats.Caches[id][stat], ResultStatVal{Val: v, Time: t, Span: 1})
- }
- if stat := "isHealthy"; filter.UseStat(stat) {
- adminDown := false
- if srv, ok := monitorConfig.TrafficServer[string(id)]; ok && enum.CacheStatusFromString(srv.Status) == enum.CacheStatusAdminDown {
- adminDown = true
- }
- v := !adminDown && combinedStates.Caches[id].IsAvailable // if the cache is missing, default to false
- stats.Caches[id][stat] = append(stats.Caches[id][stat], ResultStatVal{Val: v, Time: t, Span: 1})
- }
- if stat := "kbps"; filter.UseStat(stat) {
- v := info.Vitals.KbpsOut
- stats.Caches[id][stat] = append(stats.Caches[id][stat], ResultStatVal{Val: v, Time: t, Span: 1})
- }
- if stat := "loadAvg"; filter.UseStat(stat) {
- v := info.Vitals.LoadAvg
- stats.Caches[id][stat] = append(stats.Caches[id][stat], ResultStatVal{Val: v, Time: t, Span: 1})
- }
- if stat := "maxKbps"; filter.UseStat(stat) {
- v := info.Vitals.MaxKbpsOut
- stats.Caches[id][stat] = append(stats.Caches[id][stat], ResultStatVal{Val: v, Time: t, Span: 1})
- }
- if stat := "queryTime"; filter.UseStat(stat) {
- v := fmt.Sprintf("%d", info.RequestTime.Nanoseconds()/nsPerMs)
- stats.Caches[id][stat] = append(stats.Caches[id][stat], ResultStatVal{Val: v, Time: t, Span: 1})
- }
- if stat := "stateUrl"; filter.UseStat(stat) {
- v, err := getHealthPollingURL(id, monitorConfig)
- if err != nil {
- v = fmt.Sprintf("ERROR: %v", err) // should never happen
- }
- stats.Caches[id][stat] = append(stats.Caches[id][stat], ResultStatVal{Val: v, Time: t, Span: 1})
- }
- if stat := "status"; filter.UseStat(stat) {
- v := ""
- srv, ok := monitorConfig.TrafficServer[string(id)]
- if !ok {
- v = fmt.Sprintf("ERROR: cache not found in monitor config") // should never happen
+ for stat, statValF := range computedStats {
+ if !filter.UseStat(stat) {
+ continue
}
- v = srv.Status
- stats.Caches[id][stat] = append(stats.Caches[id][stat], ResultStatVal{Val: v, Time: t, Span: 1})
- }
- if stat := "system.astatsLoad"; filter.UseStat(stat) {
- v := info.System.AstatsLoad
- stats.Caches[id][stat] = append(stats.Caches[id][stat], ResultStatVal{Val: v, Time: t, Span: 1})
- }
- if stat := "system.configReloadRequests"; filter.UseStat(stat) {
- v := info.System.ConfigLoadRequest
- stats.Caches[id][stat] = append(stats.Caches[id][stat], ResultStatVal{Val: v, Time: t, Span: 1})
- }
- if stat := "system.configReloads"; filter.UseStat(stat) {
- v := info.System.ConfigReloads
- stats.Caches[id][stat] = append(stats.Caches[id][stat], ResultStatVal{Val: v, Time: t, Span: 1})
- }
- if stat := "system.inf.name"; filter.UseStat(stat) {
- v := info.System.InfName
- stats.Caches[id][stat] = append(stats.Caches[id][stat], ResultStatVal{Val: v, Time: t, Span: 1})
- }
- if stat := "system.inf.speed"; filter.UseStat(stat) {
- v := info.System.InfSpeed
- stats.Caches[id][stat] = append(stats.Caches[id][stat], ResultStatVal{Val: v, Time: t, Span: 1})
- }
- if stat := "system.lastReload"; filter.UseStat(stat) {
- v := info.System.LastReload
- stats.Caches[id][stat] = append(stats.Caches[id][stat], ResultStatVal{Val: v, Time: t, Span: 1})
- }
- if stat := "system.lastReloadRequest"; filter.UseStat(stat) {
- v := info.System.LastReloadRequest
- stats.Caches[id][stat] = append(stats.Caches[id][stat], ResultStatVal{Val: v, Time: t, Span: 1})
- }
- if stat := "system.proc.loadavg"; filter.UseStat(stat) {
- v := info.System.ProcLoadavg
- stats.Caches[id][stat] = append(stats.Caches[id][stat], ResultStatVal{Val: v, Time: t, Span: 1})
- }
- if stat := "system.proc.net.dev"; filter.UseStat(stat) {
- v := info.System.ProcNetDev
- stats.Caches[id][stat] = append(stats.Caches[id][stat], ResultStatVal{Val: v, Time: t, Span: 1})
+ stats.Caches[id][stat] = append(stats.Caches[id][stat], ResultStatVal{Val: statValF(resultInfo, serverInfo, serverProfile, combinedStates.Caches[id]), Time: t, Span: 1}) // combinedState will default to unavailable
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/8a7bed40/traffic_monitor/experimental/traffic_monitor/cache/data.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/cache/data.go b/traffic_monitor/experimental/traffic_monitor/cache/data.go
index 811babb..ee21907 100644
--- a/traffic_monitor/experimental/traffic_monitor/cache/data.go
+++ b/traffic_monitor/experimental/traffic_monitor/cache/data.go
@@ -36,6 +36,8 @@ type AvailableStatus struct {
Available bool
Status string
Why string
+ // UnavailableStat is the stat whose threshold made the cache unavailable. If this is the empty string, the cache is unavailable for a non-threshold reason. This exists so a poller (health, stat) won't mark an unavailable cache as available if the stat whose threshold was reached isn't available on that poller.
+ UnavailableStat string
}
// CacheAvailableStatuses is the available status of each cache.
@@ -161,7 +163,7 @@ type ResultInfo struct {
Available bool
}
-func toInfo(r Result) ResultInfo {
+func ToInfo(r Result) ResultInfo {
return ResultInfo{
ID: r.ID,
Error: r.Error,
@@ -177,7 +179,7 @@ func toInfo(r Result) ResultInfo {
func toInfos(rs []Result) []ResultInfo {
infos := make([]ResultInfo, len(rs), len(rs))
for i, r := range rs {
- infos[i] = toInfo(r)
+ infos[i] = ToInfo(r)
}
return infos
}
@@ -204,7 +206,7 @@ func pruneInfos(history []ResultInfo, limit uint64) []ResultInfo {
}
func (a ResultInfoHistory) Add(r Result, limit uint64) {
- a[r.ID] = pruneInfos(append([]ResultInfo{toInfo(r)}, a[r.ID]...), limit)
+ a[r.ID] = pruneInfos(append([]ResultInfo{ToInfo(r)}, a[r.ID]...), limit)
}
// Kbpses is the kbps values of each cache.
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/8a7bed40/traffic_monitor/experimental/traffic_monitor/enum/enum.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/enum/enum.go b/traffic_monitor/experimental/traffic_monitor/enum/enum.go
index 853236a..44184df 100644
--- a/traffic_monitor/experimental/traffic_monitor/enum/enum.go
+++ b/traffic_monitor/experimental/traffic_monitor/enum/enum.go
@@ -175,3 +175,38 @@ func CacheStatusFromString(s string) CacheStatus {
return CacheStatusInvalid
}
}
+
+// toNumeric returns a float for any numeric type, and false if the interface does not hold a numeric type.
+// This allows converting unknown numeric types (for example, from JSON) in a single line
+// TODO move to 'util' or 'tools' package?
+// TODO try to parse string stats as numbers?
+func ToNumeric(v interface{}) (float64, bool) {
+ switch i := v.(type) {
+ case uint8:
+ return float64(i), true
+ case uint16:
+ return float64(i), true
+ case uint32:
+ return float64(i), true
+ case uint64:
+ return float64(i), true
+ case int8:
+ return float64(i), true
+ case int16:
+ return float64(i), true
+ case int32:
+ return float64(i), true
+ case int64:
+ return float64(i), true
+ case float32:
+ return float64(i), true
+ case float64:
+ return float64(i), true
+ case int:
+ return float64(i), true
+ case uint:
+ return float64(i), true
+ default:
+ return 0.0, false
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/8a7bed40/traffic_monitor/experimental/traffic_monitor/health/cache_health.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/health/cache_health.go b/traffic_monitor/experimental/traffic_monitor/health/cache_health.go
index 7e9af51..15d5b9a 100644
--- a/traffic_monitor/experimental/traffic_monitor/health/cache_health.go
+++ b/traffic_monitor/experimental/traffic_monitor/health/cache_health.go
@@ -28,7 +28,8 @@ import (
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/common/log"
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/cache"
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/enum"
- traffic_ops "github.com/apache/incubator-trafficcontrol/traffic_ops/client"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/peer"
+ to "github.com/apache/incubator-trafficcontrol/traffic_ops/client"
)
func setError(newResult *cache.Result, err error) {
@@ -37,7 +38,7 @@ func setError(newResult *cache.Result, err error) {
}
// GetVitals Gets the vitals to decide health on in the right format
-func GetVitals(newResult *cache.Result, prevResult *cache.Result, mc *traffic_ops.TrafficMonitorConfigMap) {
+func GetVitals(newResult *cache.Result, prevResult *cache.Result, mc *to.TrafficMonitorConfigMap) {
if newResult.Error != nil {
log.Errorf("cache_health.GetVitals() called with an errored Result!")
return
@@ -89,7 +90,7 @@ func GetVitals(newResult *cache.Result, prevResult *cache.Result, mc *traffic_op
// inf.speed -- value looks like "10000" (without the quotes) so it is in Mbps.
// TODO JvD: Should we really be running this code every second for every cache polled????? I don't think so.
interfaceBandwidth := newResult.Astats.System.InfSpeed
- newResult.Vitals.MaxKbpsOut = int64(interfaceBandwidth)*1000 - mc.Profile[mc.TrafficServer[string(newResult.ID)].Profile].Parameters.MinFreeKbps
+ newResult.Vitals.MaxKbpsOut = int64(interfaceBandwidth) * 1000
// log.Infoln(newResult.Id, "BytesOut", newResult.Vitals.BytesOut, "BytesIn", newResult.Vitals.BytesIn, "Kbps", newResult.Vitals.KbpsOut, "max", newResult.Vitals.MaxKbpsOut)
}
@@ -123,13 +124,25 @@ func cacheCapacityKbps(result cache.Result) int64 {
return int64(result.Astats.System.InfSpeed) * kbpsInMbps
}
-// EvalCache returns whether the given cache should be marked available, and a string describing why
-func EvalCache(result cache.Result, mc *traffic_ops.TrafficMonitorConfigMap) (bool, string) {
- toServer := mc.TrafficServer[string(result.ID)]
- status := enum.CacheStatusFromString(toServer.Status)
- params := mc.Profile[toServer.Profile].Parameters
- kbpsThreshold, hasKbpsThreshold := getKbpsThreshold(params.HealthThresholdAvailableBandwidthInKbps)
- queryTimeThreshold, hasQueryTimeThreshold := getQueryThreshold(int64(params.HealthThresholdQueryTime))
+// EvalCache returns whether the given cache should be marked available, a string describing why, and which stat exceeded a threshold. The `stats` may be nil, for pollers which don't poll stats.
+// The availability of EvalCache MAY NOT be used to directly set the cache's local availability, because the threshold stats may not be part of the poller which produced the result. Rather, if the cache was previously unavailable from a threshold, it must be verified that threshold stat is in the results before setting the cache to available.
+// TODO change to return a `cache.AvailableStatus`
+func EvalCache(result cache.ResultInfo, resultStats cache.ResultStatValHistory, mc *to.TrafficMonitorConfigMap) (bool, string, string) {
+ serverInfo, ok := mc.TrafficServer[string(result.ID)]
+ if !ok {
+ log.Errorf("Cache %v missing from from Traffic Ops Monitor Config - treating as OFFLINE\n", result.ID)
+ return false, "ERROR - server missing in Traffic Ops monitor config", ""
+ }
+ serverProfile, ok := mc.Profile[serverInfo.Profile]
+ if !ok {
+ log.Errorf("Cache %v profile %v missing from from Traffic Ops Monitor Config - treating as OFFLINE\n", result.ID, serverInfo.Profile)
+ return false, "ERROR - server profile missing in Traffic Ops monitor config", ""
+ }
+
+ status := enum.CacheStatusFromString(serverInfo.Status)
+ if status == enum.CacheStatusInvalid {
+ log.Errorf("Cache %v got invalid status from Traffic Ops '%v' - treating as Reported\n", result.ID, serverInfo.Status)
+ }
availability := "available"
if !result.Available {
@@ -138,25 +151,87 @@ func EvalCache(result cache.Result, mc *traffic_ops.TrafficMonitorConfigMap) (bo
switch {
case status == enum.CacheStatusInvalid:
- log.Errorf("Cache %v got invalid status from Traffic Ops '%v' - treating as OFFLINE\n", result.ID, toServer.Status)
- return false, getEventDescription(status, availability+"; invalid status")
+ log.Errorf("Cache %v got invalid status from Traffic Ops '%v' - treating as OFFLINE\n", result.ID, serverInfo.Status)
+ return false, getEventDescription(status, availability+"; invalid status"), ""
case status == enum.CacheStatusAdminDown:
- return false, getEventDescription(status, availability)
+ return false, getEventDescription(status, availability), ""
case status == enum.CacheStatusOffline:
- log.Errorf("Cache %v set to OFFLINE, but still polled\n", result.ID)
- return false, getEventDescription(status, availability)
+ log.Errorf("Cache %v set to offline, but still polled\n", result.ID)
+ return false, getEventDescription(status, availability), ""
case status == enum.CacheStatusOnline:
- return true, getEventDescription(status, availability)
+ return true, getEventDescription(status, availability), ""
case result.Error != nil:
- return false, getEventDescription(status, fmt.Sprintf("%v", result.Error))
- case result.Vitals.LoadAvg > params.HealthThresholdLoadAvg && params.HealthThresholdLoadAvg != 0:
- return false, getEventDescription(status, fmt.Sprintf("loadavg too high (%.5f > %.5f)", result.Vitals.LoadAvg, params.HealthThresholdLoadAvg))
- case hasKbpsThreshold && cacheCapacityKbps(result)-result.Vitals.KbpsOut < kbpsThreshold:
- return false, getEventDescription(status, fmt.Sprintf("availableBandwidthInKbps too low (%d < %d)", cacheCapacityKbps(result)-result.Vitals.KbpsOut, kbpsThreshold))
- case hasQueryTimeThreshold && result.RequestTime > queryTimeThreshold:
- return false, getEventDescription(status, fmt.Sprintf("queryTime too high (%.5f > %.5f)", float64(result.RequestTime.Nanoseconds())/1e6, float64(queryTimeThreshold.Nanoseconds())/1e6))
+ return false, getEventDescription(status, fmt.Sprintf("%v", result.Error)), ""
+ }
+
+ computedStats := cache.ComputedStats()
+
+ for stat, threshold := range serverProfile.Parameters.Thresholds {
+ resultStat := interface{}(nil)
+ if computedStatF, ok := computedStats[stat]; ok {
+ dummyCombinedstate := peer.IsAvailable{} // the only stats which use combinedState are things like isAvailable, which don't make sense to ever be thresholds.
+ resultStat = computedStatF(result, serverInfo, serverProfile, dummyCombinedstate)
+ } else {
+ if resultStats == nil {
+ continue
+ }
+ resultStatHistory, ok := resultStats[stat]
+ if !ok {
+ continue
+ }
+ if len(resultStatHistory) < 1 {
+ continue
+ }
+ resultStat = resultStatHistory[0].Val
+ }
+
+ resultStatNum, ok := enum.ToNumeric(resultStat)
+ if !ok {
+ log.Errorf("health.EvalCache threshold stat %s was not a number: %v", stat, resultStat)
+ continue
+ }
+
+ if !InThreshold(threshold, resultStatNum) {
+ return false, getEventDescription(status, ExceedsThresholdMsg(stat, threshold, resultStatNum)), stat
+ }
+ }
+
+ return result.Available, getEventDescription(status, availability), ""
+}
+
+// ExceedsThresholdMsg returns a human-readable message for why the given value exceeds the threshold. It does NOT check whether the value actually exceeds the threshold; call `InThreshold` to check first.
+func ExceedsThresholdMsg(stat string, threshold to.HealthThreshold, val float64) string {
+ switch threshold.Comparator {
+ case "=":
+ return fmt.Sprintf("%s not equal (%f != %f)", stat, val, threshold.Val)
+ case ">":
+ return fmt.Sprintf("%s too low (%f < %f)", stat, val, threshold.Val)
+ case "<":
+ return fmt.Sprintf("%s too high (%f > %f)", stat, val, threshold.Val)
+ case ">=":
+ return fmt.Sprintf("%s too low (%f <= %f)", stat, val, threshold.Val)
+ case "<=":
+ return fmt.Sprintf("%s too high (%f >= %f)", stat, val, threshold.Val)
+ default:
+ return fmt.Sprintf("ERROR: Invalid Threshold: %+v", threshold)
+ }
+}
+
+func InThreshold(threshold to.HealthThreshold, val float64) bool {
+ switch threshold.Comparator {
+ case "=":
+ return val == threshold.Val
+ case ">":
+ return val > threshold.Val
+ case "<":
+ return val < threshold.Val
+ case ">=":
+ return val >= threshold.Val
+ case "<=":
+ return val <= threshold.Val
default:
- return result.Available, getEventDescription(status, availability)
+ log.Errorf("Invalid Threshold: %+v", threshold)
+ return true // for safety, if a threshold somehow gets corrupted, don't start marking caches down.
}
}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/8a7bed40/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go b/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
index c4385f3..2f42d74 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
@@ -902,7 +902,7 @@ func createCacheStatuses(
) map[enum.CacheName]CacheStatus {
conns := createCacheConnections(statResultHistory)
statii := map[enum.CacheName]CacheStatus{}
- localCacheStatus := localCacheStatusThreadsafe.Get()
+ localCacheStatus := localCacheStatusThreadsafe.Get().Copy() // TODO test whether copy is necessary
maxKbpses := statMaxKbpses.Get()
for cacheName, cacheType := range cacheTypes {
@@ -1162,39 +1162,6 @@ type StatSummaryStat struct {
EndTime int64 `json:"endTime"`
}
-// toNumeric returns a float for any numeric type, and false if the interface does not hold a numeric type.
-// This allows converting unknown numeric types (for example, from JSON) in a single line
-func toNumeric(v interface{}) (float64, bool) {
- switch i := v.(type) {
- case uint8:
- return float64(i), true
- case uint16:
- return float64(i), true
- case uint32:
- return float64(i), true
- case uint64:
- return float64(i), true
- case int8:
- return float64(i), true
- case int16:
- return float64(i), true
- case int32:
- return float64(i), true
- case int64:
- return float64(i), true
- case float32:
- return float64(i), true
- case float64:
- return float64(i), true
- case int:
- return float64(i), true
- case uint:
- return float64(i), true
- default:
- return 0.0, false
- }
-}
-
func createStatSummary(statResultHistory cache.ResultStatHistory, filter cache.Filter, params url.Values) StatSummary {
statPrefix := "ats."
ss := StatSummary{
@@ -1217,8 +1184,8 @@ func createStatSummary(statResultHistory cache.ResultStatHistory, filter cache.F
msPerNs := int64(1000000)
ssStat.StartTime = time.Time(statHistory[len(statHistory)-1].Time).UnixNano() / msPerNs
ssStat.EndTime = time.Time(statHistory[0].Time).UnixNano() / msPerNs
- oldestVal, isOldestValNumeric := toNumeric(statHistory[len(statHistory)-1].Val)
- newestVal, isNewestValNumeric := toNumeric(statHistory[0].Val)
+ oldestVal, isOldestValNumeric := enum.ToNumeric(statHistory[len(statHistory)-1].Val)
+ newestVal, isNewestValNumeric := enum.ToNumeric(statHistory[0].Val)
if !isOldestValNumeric || !isNewestValNumeric {
continue // skip non-numeric stats
}
@@ -1227,7 +1194,7 @@ func createStatSummary(statResultHistory cache.ResultStatHistory, filter cache.F
ssStat.High = newestVal
ssStat.Low = newestVal
for _, val := range statHistory {
- fVal, ok := toNumeric(val.Val)
+ fVal, ok := enum.ToNumeric(val.Val)
if !ok {
continue // skip non-numeric stats. TODO warn about stat history containing different types?
}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/8a7bed40/traffic_monitor/experimental/traffic_monitor/manager/healthresult.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/healthresult.go b/traffic_monitor/experimental/traffic_monitor/manager/healthresult.go
index ed9f9fc..f99155d 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/healthresult.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/healthresult.go
@@ -87,9 +87,9 @@ func StartHealthResultManager(
errorCount threadsafe.Uint,
cfg config.Config,
events threadsafe.Events,
-) (DurationMapThreadsafe, threadsafe.CacheAvailableStatus, threadsafe.ResultHistory) {
+ localCacheStatus threadsafe.CacheAvailableStatus,
+) (DurationMapThreadsafe, threadsafe.ResultHistory) {
lastHealthDurations := NewDurationMapThreadsafe()
- localCacheStatus := threadsafe.NewCacheAvailableStatus()
healthHistory := threadsafe.NewResultHistory()
go healthResultManagerListen(
cacheHealthChan,
@@ -106,7 +106,7 @@ func StartHealthResultManager(
localCacheStatus,
cfg,
)
- return lastHealthDurations, localCacheStatus, healthHistory
+ return lastHealthDurations, healthHistory
}
func healthResultManagerListen(
@@ -127,6 +127,27 @@ func healthResultManagerListen(
lastHealthEndTimes := map[enum.CacheName]time.Time{}
// This reads at least 1 value from the cacheHealthChan. Then, we loop, and try to read from the channel some more. If there's nothing to read, we hit `default` and process. If there is stuff to read, we read it, then inner-loop trying to read more. If we're continuously reading and the channel is never empty, and we hit the tick time, process anyway even though the channel isn't empty, to prevent never processing (starvation).
var ticker *time.Ticker
+
+ process := func(results []cache.Result) {
+ processHealthResult(
+ cacheHealthChan,
+ toData,
+ localStates,
+ lastHealthDurations,
+ monitorConfig,
+ peerStates,
+ combinedStates,
+ fetchCount,
+ errorCount,
+ events,
+ localCacheStatus,
+ lastHealthEndTimes,
+ healthHistory,
+ results,
+ cfg,
+ )
+ }
+
for {
var results []cache.Result
results = append(results, <-cacheHealthChan)
@@ -139,46 +160,14 @@ func healthResultManagerListen(
select {
case <-ticker.C:
log.Warnf("Health Result Manager flushing queued results\n")
- processHealthResult(
- cacheHealthChan,
- toData,
- localStates,
- lastHealthDurations,
- monitorConfig,
- peerStates,
- combinedStates,
- fetchCount,
- errorCount,
- events,
- localCacheStatus,
- lastHealthEndTimes,
- healthHistory,
- results,
- cfg,
- )
+ process(results)
break innerLoop
default:
select {
case r := <-cacheHealthChan:
results = append(results, r)
default:
- processHealthResult(
- cacheHealthChan,
- toData,
- localStates,
- lastHealthDurations,
- monitorConfig,
- peerStates,
- combinedStates,
- fetchCount,
- errorCount,
- events,
- localCacheStatus,
- lastHealthEndTimes,
- healthHistory,
- results,
- cfg,
- )
+ process(results)
break innerLoop
}
}
@@ -207,6 +196,13 @@ func processHealthResult(
if len(results) == 0 {
return
}
+ defer func() {
+ for _, r := range results {
+ log.Debugf("poll %v %v finish\n", r.PollID, time.Now())
+ r.PollFinished <- r.PollID
+ }
+ }()
+
toDataCopy := toData.Get() // create a copy, so the same data used for all processing of this cache health result
localCacheStatus := localCacheStatusThreadsafe.Get().Copy()
monitorConfigCopy := monitorConfig.Get()
@@ -232,20 +228,30 @@ func processHealthResult(
healthHistoryCopy[healthResult.ID] = pruneHistory(append([]cache.Result{healthResult}, healthHistoryCopy[healthResult.ID]...), maxHistory)
- isAvailable, whyAvailable := health.EvalCache(healthResult, &monitorConfigCopy)
+ isAvailable, whyAvailable, unavailableStat := health.EvalCache(cache.ToInfo(healthResult), nil, &monitorConfigCopy)
+ whyAvailable += "(healthpoll)" // debug
if available, ok := localStates.GetCache(healthResult.ID); !ok || available.IsAvailable != isAvailable {
- log.Infof("Changing state for %s was: %t now: %t because %s error: %v", healthResult.ID, prevResult.Available, isAvailable, whyAvailable, healthResult.Error)
- events.Add(health.Event{Time: healthResult.Time, Unix: healthResult.Time.Unix(), Description: whyAvailable, Name: healthResult.ID.String(), Hostname: healthResult.ID.String(), Type: toDataCopy.ServerTypes[healthResult.ID].String(), Available: isAvailable})
+ log.Infof("Changing state for %s was: %t now: %t because %s error: %v", healthResult.ID, available.IsAvailable, isAvailable, whyAvailable, healthResult.Error)
+ events.Add(health.Event{Time: time.Now(), Description: whyAvailable, Name: string(healthResult.ID), Hostname: string(healthResult.ID), Type: toDataCopy.ServerTypes[healthResult.ID].String(), Available: isAvailable})
+ }
+
+ // if the cache is now Available, and was previously unavailable due to a threshold, make sure this poller contains the stat which exceeded the threshold.
+ if previousStatus, hasPreviousStatus := localCacheStatus[healthResult.ID]; isAvailable && hasPreviousStatus && !previousStatus.Available && previousStatus.UnavailableStat != "" {
+ if !resultHasStat(previousStatus.UnavailableStat, healthResult) {
+ // TODO determine if it's ok to add the result data (but not availability). Or will making them not align cause issues?
+ continue
+ }
}
localCacheStatus[healthResult.ID] = cache.AvailableStatus{
- Available: isAvailable,
- Status: monitorConfigCopy.TrafficServer[string(healthResult.ID)].Status,
- Why: whyAvailable,
+ Available: isAvailable,
+ Status: monitorConfigCopy.TrafficServer[string(healthResult.ID)].Status,
+ Why: whyAvailable,
+ UnavailableStat: unavailableStat,
} // TODO move within localStates?
localStates.SetCache(healthResult.ID, peer.IsAvailable{IsAvailable: isAvailable})
}
- calculateDeliveryServiceState(toDataCopy.DeliveryServiceServers, localStates)
+ CalculateDeliveryServiceState(toDataCopy.DeliveryServiceServers, localStates)
healthHistory.Set(healthHistoryCopy)
localCacheStatusThreadsafe.Set(localCacheStatus)
// TODO determine if we should combineCrStates() here
@@ -257,15 +263,25 @@ func processHealthResult(
lastHealthDurations[healthResult.ID] = d
}
lastHealthEndTimes[healthResult.ID] = time.Now()
-
- log.Debugf("poll %v %v finish\n", healthResult.PollID, time.Now())
- healthResult.PollFinished <- healthResult.PollID
}
lastHealthDurationsThreadsafe.Set(lastHealthDurations)
}
+// resultHasStat returns whether the given stat is in the Result.
+// TODO move to cache?
+func resultHasStat(stat string, result cache.Result) bool {
+ computedStats := cache.ComputedStats()
+ if _, ok := computedStats[stat]; ok {
+ return true // health poll has all computed stats
+ }
+ if _, ok := result.Astats.Ats[stat]; ok {
+ return true
+ }
+ return false
+}
+
// calculateDeliveryServiceState calculates the state of delivery services from the new cache state data `cacheState` and the CRConfig data `deliveryServiceServers` and puts the calculated state in the outparam `deliveryServiceStates`
-func calculateDeliveryServiceState(deliveryServiceServers map[enum.DeliveryServiceName][]enum.CacheName, states peer.CRStatesThreadsafe) {
+func CalculateDeliveryServiceState(deliveryServiceServers map[enum.DeliveryServiceName][]enum.CacheName, states peer.CRStatesThreadsafe) {
deliveryServices := states.GetDeliveryServices()
for deliveryServiceName, deliveryServiceState := range deliveryServices {
if _, ok := deliveryServiceServers[deliveryServiceName]; !ok {
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/8a7bed40/traffic_monitor/experimental/traffic_monitor/manager/manager.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/manager.go b/traffic_monitor/experimental/traffic_monitor/manager/manager.go
index 26ebb37..dba10ad 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/manager.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/manager.go
@@ -102,16 +102,17 @@ func Start(opsConfigFile string, cfg config.Config, staticAppData StaticAppData)
staticAppData,
)
- combinedStates := StartPeerManager(
+ combinedStates, events := StartPeerManager(
peerHandler.ResultChannel,
localStates,
peerStates,
events,
- cfg.PeerOptimistic,
+ cfg.PeerOptimistic, // TODO remove
toData,
+ cfg,
)
- statInfoHistory, statResultHistory, statMaxKbpses, _, lastKbpsStats, dsStats, unpolledCaches := StartStatHistoryManager(
+ statInfoHistory, statResultHistory, statMaxKbpses, _, lastKbpsStats, dsStats, unpolledCaches, localCacheStatus := StartStatHistoryManager(
cacheStatHandler.ResultChannel,
localStates,
combinedStates,
@@ -120,9 +121,10 @@ func Start(opsConfigFile string, cfg config.Config, staticAppData StaticAppData)
errorCount,
cfg,
monitorConfig,
+ events,
)
- lastHealthDurations, localCacheStatus, healthHistory := StartHealthResultManager(
+ lastHealthDurations, healthHistory := StartHealthResultManager(
cacheHealthHandler.ResultChannel,
toData,
localStates,
@@ -133,6 +135,7 @@ func Start(opsConfigFile string, cfg config.Config, staticAppData StaticAppData)
errorCount,
cfg,
events,
+ localCacheStatus,
)
StartOpsConfigManager(
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/8a7bed40/traffic_monitor/experimental/traffic_monitor/manager/peer.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/peer.go b/traffic_monitor/experimental/traffic_monitor/manager/peer.go
index 2f8cbdb..13b8411 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/peer.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/peer.go
@@ -27,6 +27,7 @@ import (
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/common/log"
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/common/util"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/config"
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/enum"
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/health"
"github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/peer"
@@ -42,7 +43,8 @@ func StartPeerManager(
events threadsafe.Events,
peerOptimistic bool,
toData todata.TODataThreadsafe,
-) peer.CRStatesThreadsafe {
+ cfg config.Config,
+) (peer.CRStatesThreadsafe, threadsafe.Events) {
combinedStates := peer.NewCRStatesThreadsafe()
overrideMap := map[enum.CacheName]bool{}
@@ -54,7 +56,7 @@ func StartPeerManager(
peerResult.PollFinished <- peerResult.PollID
}
}()
- return combinedStates
+ return combinedStates, events
}
func comparePeerState(events threadsafe.Events, result peer.Result, peerStates peer.CRStatesPeersThreadsafe) {
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/8a7bed40/traffic_monitor/experimental/traffic_monitor/manager/stathistory.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/stathistory.go b/traffic_monitor/experimental/traffic_monitor/manager/stathistory.go
index d28713d..bf2d1dc 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/stathistory.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/stathistory.go
@@ -67,7 +67,8 @@ func StartStatHistoryManager(
errorCount threadsafe.Uint,
cfg config.Config,
monitorConfig TrafficMonitorConfigMapThreadsafe,
-) (threadsafe.ResultInfoHistory, threadsafe.ResultStatHistory, threadsafe.CacheKbpses, DurationMapThreadsafe, threadsafe.LastStats, threadsafe.DSStatsReader, threadsafe.UnpolledCaches) {
+ events threadsafe.Events,
+) (threadsafe.ResultInfoHistory, threadsafe.ResultStatHistory, threadsafe.CacheKbpses, DurationMapThreadsafe, threadsafe.LastStats, threadsafe.DSStatsReader, threadsafe.UnpolledCaches, threadsafe.CacheAvailableStatus) {
statInfoHistory := threadsafe.NewResultInfoHistory()
statResultHistory := threadsafe.NewResultStatHistory()
statMaxKbpses := threadsafe.NewCacheKbpses()
@@ -77,12 +78,13 @@ func StartStatHistoryManager(
dsStats := threadsafe.NewDSStats()
unpolledCaches := threadsafe.NewUnpolledCaches()
tickInterval := cfg.StatFlushInterval
+ localCacheStatus := threadsafe.NewCacheAvailableStatus()
precomputedData := map[enum.CacheName]cache.PrecomputedData{}
lastResults := map[enum.CacheName]cache.Result{}
process := func(results []cache.Result) {
- processStatResults(results, statInfoHistory, statResultHistory, statMaxKbpses, combinedStates.Get(), lastStats, toData.Get(), errorCount, dsStats, lastStatEndTimes, lastStatDurations, unpolledCaches, monitorConfig.Get(), precomputedData, lastResults)
+ processStatResults(results, statInfoHistory, statResultHistory, statMaxKbpses, combinedStates.Get(), lastStats, toData.Get(), errorCount, dsStats, lastStatEndTimes, lastStatDurations, unpolledCaches, monitorConfig.Get(), precomputedData, lastResults, localStates, events, localCacheStatus)
}
go func() {
@@ -118,7 +120,7 @@ func StartStatHistoryManager(
}
}
}()
- return statInfoHistory, statResultHistory, statMaxKbpses, lastStatDurations, lastStats, &dsStats, unpolledCaches
+ return statInfoHistory, statResultHistory, statMaxKbpses, lastStatDurations, lastStats, &dsStats, unpolledCaches, localCacheStatus
}
// processStatResults processes the given results, creating and setting DSStats, LastStats, and other stats. Note this is NOT threadsafe, and MUST NOT be called from multiple threads.
@@ -138,14 +140,27 @@ func processStatResults(
mc to.TrafficMonitorConfigMap,
precomputedData map[enum.CacheName]cache.PrecomputedData,
lastResults map[enum.CacheName]cache.Result,
+ localStates peer.CRStatesThreadsafe,
+ events threadsafe.Events,
+ localCacheStatusThreadsafe threadsafe.CacheAvailableStatus,
) {
+ if len(results) == 0 {
+ return
+ }
+ defer func() {
+ for _, r := range results {
+ // log.Debugf("poll %v %v statfinish\n", result.PollID, endTime)
+ r.PollFinished <- r.PollID
+ }
+ }()
// setting the statHistory could be put in a goroutine concurrent with `ds.CreateStats`, if it were slow
statInfoHistory := statInfoHistoryThreadsafe.Get().Copy()
statResultHistory := statResultHistoryThreadsafe.Get().Copy()
statMaxKbpses := statMaxKbpsesThreadsafe.Get().Copy()
+ localCacheStatus := localCacheStatusThreadsafe.Get().Copy()
- for _, result := range results {
+ for i, result := range results {
maxStats := uint64(mc.Profile[mc.TrafficServer[string(result.ID)].Profile].Parameters.HistoryCount)
if maxStats < 1 {
log.Warnf("processStatResults got history count %v for %v, setting to 1\n", maxStats, result.ID)
@@ -153,8 +168,13 @@ func processStatResults(
}
// TODO determine if we want to add results with errors, or just print the errors now and don't add them.
- if lastResult, ok := lastResults[result.ID]; ok {
+ if lastResult, ok := lastResults[result.ID]; ok && result.Error == nil {
health.GetVitals(&result, &lastResult, &mc) // TODO precompute
+ if result.Error == nil {
+ results[i] = result
+ } else {
+ log.Errorf("stat poll getting vitals for %v: %v\n", result.ID, result.Error)
+ }
}
statInfoHistory.Add(result, maxStats)
statResultHistory.Add(result, maxStats)
@@ -193,6 +213,35 @@ func processStatResults(
lastStats.Set(newLastStats)
}
+ // TODO test
+ // TODO abstract setting availability logic (duplicated in healthresult.go)
+ for _, result := range results {
+ isAvailable, whyAvailable, unavailableStat := health.EvalCache(cache.ToInfo(result), statResultHistory[result.ID], &mc)
+ whyAvailable += "(statpoll)" // debug
+
+ if available, ok := localStates.GetCache(result.ID); !ok || available.IsAvailable != isAvailable {
+ log.Infof("Changing state for %s was: %t now: %t because %s error: %v", result.ID, available.IsAvailable, isAvailable, whyAvailable, result.Error)
+ events.Add(health.Event{Time: time.Now(), Description: whyAvailable, Name: string(result.ID), Hostname: string(result.ID), Type: toData.ServerTypes[result.ID].String(), Available: isAvailable})
+ }
+
+ // if the cache is now Available, and was previously unavailable due to a threshold, make sure this poller contains the stat which exceeded the threshold.
+ if previousStatus, hasPreviousStatus := localCacheStatus[result.ID]; isAvailable && hasPreviousStatus && !previousStatus.Available && previousStatus.UnavailableStat != "" {
+ if !resultHasStat(previousStatus.UnavailableStat, result) {
+ // TODO determine if it's ok to add the result data (but not availability). Or will making them not align cause issues?
+ continue
+ }
+ }
+ localCacheStatus[result.ID] = cache.AvailableStatus{
+ Available: isAvailable,
+ Status: mc.TrafficServer[string(result.ID)].Status,
+ Why: whyAvailable,
+ UnavailableStat: unavailableStat,
+ } // TODO move within localStates?
+ localStates.SetCache(result.ID, peer.IsAvailable{IsAvailable: isAvailable})
+ CalculateDeliveryServiceState(toData.DeliveryServiceServers, localStates)
+ localCacheStatusThreadsafe.Set(localCacheStatus)
+ }
+
endTime := time.Now()
lastStatDurations := lastStatDurationsThreadsafe.Get().Copy()
for _, result := range results {
@@ -201,9 +250,6 @@ func processStatResults(
lastStatDurations[result.ID] = d
}
lastStatEndTimes[result.ID] = endTime
-
- // log.Debugf("poll %v %v statfinish\n", result.PollID, endTime)
- result.PollFinished <- result.PollID
}
lastStatDurationsThreadsafe.Set(lastStatDurations)
unpolledCaches.SetPolled(results, lastStats.Get())
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/8a7bed40/traffic_monitor/experimental/traffic_monitor/threadsafe/events.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/threadsafe/events.go b/traffic_monitor/experimental/traffic_monitor/threadsafe/events.go
index fe78ba5..9cd0320 100644
--- a/traffic_monitor/experimental/traffic_monitor/threadsafe/events.go
+++ b/traffic_monitor/experimental/traffic_monitor/threadsafe/events.go
@@ -57,13 +57,14 @@ func (o *Events) Get() []health.Event {
func (o *Events) Add(e health.Event) {
// host="hostname", type=EDGE, available=true, msg="REPORTED - available"
log.Eventf(e.Time, "host=\"%s\", type=%s, available=%t, msg=\"%s\"", e.Hostname, e.Type, e.Available, e.Description)
+ o.m.Lock() // TODO test removing
events := copyEvents(*o.events)
e.Index = *o.nextIndex
events = append([]health.Event{e}, events...)
if len(events) > int(o.max) {
events = (events)[:o.max-1]
}
- o.m.Lock()
+ // o.m.Lock()
*o.events = events
*o.nextIndex++
o.m.Unlock()
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/8a7bed40/traffic_ops/client/traffic_monitor_config.go
----------------------------------------------------------------------
diff --git a/traffic_ops/client/traffic_monitor_config.go b/traffic_ops/client/traffic_monitor_config.go
index cf1208b..f5ece4d 100644
--- a/traffic_ops/client/traffic_monitor_config.go
+++ b/traffic_ops/client/traffic_monitor_config.go
@@ -75,14 +75,86 @@ type TMProfile struct {
}
// TMParameters ...
+// TODO change TO to return this struct, so a custom UnmarshalJSON isn't necessary.
type TMParameters struct {
- HealthConnectionTimeout int `json:"health.connection.timeout"`
- HealthPollingURL string `json:"health.polling.url"`
- HealthThresholdQueryTime int `json:"health.threshold.queryTime"`
- HistoryCount int `json:"history.count"`
- HealthThresholdAvailableBandwidthInKbps string `json:"health.threshold.availableBandwidthInKbps"`
- HealthThresholdLoadAvg float64 `json:"health.threshold.loadavg,string"`
- MinFreeKbps int64
+ HealthConnectionTimeout int `json:"health.connection.timeout"`
+ HealthPollingURL string `json:"health.polling.url"`
+ HistoryCount int `json:"history.count"`
+ MinFreeKbps int64
+ Thresholds map[string]HealthThreshold `json:"health_threshold"`
+}
+
+const DefaultHealthThresholdComparator = "<"
+
+type HealthThreshold struct {
+ Val float64
+ Comparator string // TODO change to enum?
+}
+
+// strToThreshold takes a string like ">=42" and returns a HealthThreshold with a Val of `42` and a Comparator of `">="`. If no comparator exists, `DefaultHealthThresholdComparator` is used. If the string is not of the form "(>|<|)(=|)\d+" an error is returned
+func strToThreshold(s string) (HealthThreshold, error) {
+ comparators := []string{"=", ">", "<", ">=", "<="}
+ for _, comparator := range comparators {
+ if strings.HasPrefix(s, comparator) {
+ valStr := s[len(comparator):]
+ val, err := strconv.ParseFloat(valStr, 64)
+ if err != nil {
+ return HealthThreshold{}, fmt.Errorf("invalid threshold: NaN")
+ }
+ return HealthThreshold{Val: val, Comparator: comparator}, nil
+ }
+ }
+ val, err := strconv.ParseFloat(s, 64)
+ if err != nil {
+ return HealthThreshold{}, fmt.Errorf("invalid threshold: NaN")
+ }
+ return HealthThreshold{Val: val, Comparator: DefaultHealthThresholdComparator}, nil
+}
+
+func (params *TMParameters) UnmarshalJSON(bytes []byte) (err error) {
+ raw := map[string]interface{}{}
+ if err := json.Unmarshal(bytes, &raw); err != nil {
+ return err
+ }
+
+ if vi, ok := raw["health.connection.timeout"]; ok {
+ if v, ok := vi.(float64); !ok {
+ return fmt.Errorf("Unmarshalling TMParameters health.connection.timeout expected integer, got %v", vi)
+ } else {
+ params.HealthConnectionTimeout = int(v)
+ }
+ }
+
+ if vi, ok := raw["health.polling.url"]; ok {
+ if v, ok := vi.(string); !ok {
+ return fmt.Errorf("Unmarshalling TMParameters health.polling.url expected string, got %v", vi)
+ } else {
+ params.HealthPollingURL = v
+ }
+ }
+
+ if vi, ok := raw["history.count"]; ok {
+ if v, ok := vi.(float64); !ok {
+ return fmt.Errorf("Unmarshalling TMParameters history.count expected integer, got %v", vi)
+ } else {
+ params.HistoryCount = int(v)
+ }
+ }
+
+ params.Thresholds = map[string]HealthThreshold{}
+ thresholdPrefix := "health.threshold."
+ for k, v := range raw {
+ if strings.HasPrefix(k, thresholdPrefix) {
+ stat := k[len(thresholdPrefix):]
+ vStr := fmt.Sprintf("%v", v) // allows string or numeric JSON types. TODO check if a type switch is faster.
+ if t, err := strToThreshold(vStr); err != nil {
+ return fmt.Errorf("Unmarshalling TMParameters `health.threshold.` parameter value not of the form `(>|)(=|)\\d+`: stat '%s' value '%v'", k, v)
+ } else {
+ params.Thresholds[stat] = t
+ }
+ }
+ }
+ return nil
}
// TrafficMonitorConfigMap ...
@@ -146,14 +218,8 @@ func trafficMonitorTransformToMap(tmConfig *TrafficMonitorConfig) (*TrafficMonit
}
for _, profile := range tmConfig.Profiles {
- bwThresholdString := profile.Parameters.HealthThresholdAvailableBandwidthInKbps
- if strings.HasPrefix(bwThresholdString, ">") {
- var err error
- profile.Parameters.MinFreeKbps, err = strconv.ParseInt(bwThresholdString[1:len(bwThresholdString)], 10, 64)
- if err != nil {
- return nil, err
- }
- }
+ bwThreshold := profile.Parameters.Thresholds["availableBandwidthInKbps"]
+ profile.Parameters.MinFreeKbps = int64(bwThreshold.Val)
tm.Profile[profile.Name] = profile
}