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/30 15:29:14 UTC
[08/19] incubator-trafficcontrol git commit: Move TM2 to
trafficcontrol/traffic_monitor_golang
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/594b8517/traffic_monitor_golang/traffic_monitor/deliveryservice/stat.go
----------------------------------------------------------------------
diff --git a/traffic_monitor_golang/traffic_monitor/deliveryservice/stat.go b/traffic_monitor_golang/traffic_monitor/deliveryservice/stat.go
new file mode 100644
index 0000000..9efd2c9
--- /dev/null
+++ b/traffic_monitor_golang/traffic_monitor/deliveryservice/stat.go
@@ -0,0 +1,358 @@
+package deliveryservice
+
+/*
+ * 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"
+ "time"
+
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor_golang/common/log"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor_golang/common/util"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor_golang/traffic_monitor/cache"
+ dsdata "github.com/apache/incubator-trafficcontrol/traffic_monitor_golang/traffic_monitor/deliveryservicedata"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor_golang/traffic_monitor/enum"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor_golang/traffic_monitor/health"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor_golang/traffic_monitor/peer"
+ todata "github.com/apache/incubator-trafficcontrol/traffic_monitor_golang/traffic_monitor/trafficopsdata"
+ to "github.com/apache/incubator-trafficcontrol/traffic_ops/client"
+)
+
+// TODO remove 'ds' and 'stat' from names
+
+func setStaticData(dsStats dsdata.Stats, dsServers map[enum.DeliveryServiceName][]enum.CacheName) dsdata.Stats {
+ for ds, stat := range dsStats.DeliveryService {
+ stat.CommonStats.CachesConfiguredNum.Value = int64(len(dsServers[ds]))
+ dsStats.DeliveryService[ds] = stat // TODO consider changing dsStats.DeliveryService[ds] to a pointer so this kind of thing isn't necessary; possibly more performant, as well
+ }
+ return dsStats
+}
+
+func addAvailableData(dsStats dsdata.Stats, crStates peer.Crstates, serverCachegroups map[enum.CacheName]enum.CacheGroupName, serverDs map[enum.CacheName][]enum.DeliveryServiceName, serverTypes map[enum.CacheName]enum.CacheType, precomputed map[enum.CacheName]cache.PrecomputedData) (dsdata.Stats, error) {
+ for cache, available := range crStates.Caches {
+ cacheGroup, ok := serverCachegroups[cache]
+ if !ok {
+ log.Infof("CreateStats not adding availability data for '%s': not found in Cachegroups\n", cache)
+ continue
+ }
+ deliveryServices, ok := serverDs[cache]
+ if !ok {
+ log.Infof("CreateStats not adding availability data for '%s': not found in DeliveryServices\n", cache)
+ continue
+ }
+ cacheType, ok := serverTypes[cache]
+ if !ok {
+ log.Infof("CreateStats not adding availability data for '%s': not found in Server Types\n", cache)
+ continue
+ }
+
+ for _, deliveryService := range deliveryServices {
+ if deliveryService == "" {
+ log.Errorln("EMPTY addAvailableData DS") // various bugs in other functions can cause this - this will help identify and debug them.
+ continue
+ }
+
+ stat, ok := dsStats.DeliveryService[deliveryService]
+ if !ok {
+ log.Infof("CreateStats not adding availability data for '%s': not found in Stats\n", cache)
+ continue // TODO log warning? Error?
+ }
+
+ if available.IsAvailable {
+ stat.CommonStats.IsAvailable.Value = true
+ stat.CommonStats.IsHealthy.Value = true
+ stat.CommonStats.CachesAvailableNum.Value++
+ cacheGroupStats := stat.CacheGroups[cacheGroup]
+ cacheGroupStats.IsAvailable.Value = true
+ stat.CacheGroups[cacheGroup] = cacheGroupStats
+ stat.TotalStats.IsAvailable.Value = true
+ typeStats := stat.Types[cacheType]
+ typeStats.IsAvailable.Value = true
+ stat.Types[cacheType] = typeStats
+ }
+
+ // TODO fix nested ifs
+ if pc, ok := precomputed[cache]; ok {
+ if pc.Reporting {
+ stat.CommonStats.CachesReporting[cache] = true
+ } else {
+ log.Debugf("no reporting %v %v\n", cache, deliveryService)
+ }
+ } else {
+ log.Debugf("no result for %v %v\n", cache, deliveryService)
+ }
+
+ 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
+}
+
+func newLastDSStat() dsdata.LastDSStat {
+ return dsdata.LastDSStat{
+ CacheGroups: map[enum.CacheGroupName]dsdata.LastStatsData{},
+ Type: map[enum.CacheType]dsdata.LastStatsData{},
+ Caches: map[enum.CacheName]dsdata.LastStatsData{},
+ }
+}
+
+// BytesPerKilobit is the number of bytes in a kilobit.
+const BytesPerKilobit = 125
+
+func addLastStat(lastData dsdata.LastStatData, newStat int64, newStatTime time.Time) (dsdata.LastStatData, error) {
+ if newStat == lastData.Stat {
+ return lastData, nil
+ }
+
+ if newStat < lastData.Stat {
+ return lastData, fmt.Errorf("new stat '%d'@'%v' value less than last stat '%d'@'%v'", lastData.Stat, lastData.Time, newStat, newStatTime)
+ }
+
+ if newStatTime.Before(lastData.Time) {
+ return lastData, fmt.Errorf("new stat '%d'@'%v' time less than last stat '%d'@'%v'", lastData.Stat, lastData.Time, newStat, newStatTime)
+ }
+
+ if lastData.Stat != 0 {
+ lastData.PerSec = float64(newStat-lastData.Stat) / newStatTime.Sub(lastData.Time).Seconds()
+ }
+
+ lastData.Stat = newStat
+ lastData.Time = newStatTime
+ return lastData, nil
+}
+
+func addLastStats(lastData dsdata.LastStatsData, newStats dsdata.StatCacheStats, newStatsTime time.Time) (dsdata.LastStatsData, error) {
+ errs := []error{nil, nil, nil, nil, nil}
+ lastData.Bytes, errs[0] = addLastStat(lastData.Bytes, newStats.OutBytes.Value, newStatsTime)
+ lastData.Status2xx, errs[1] = addLastStat(lastData.Status2xx, newStats.Status2xx.Value, newStatsTime)
+ lastData.Status3xx, errs[2] = addLastStat(lastData.Status3xx, newStats.Status3xx.Value, newStatsTime)
+ lastData.Status4xx, errs[3] = addLastStat(lastData.Status4xx, newStats.Status4xx.Value, newStatsTime)
+ lastData.Status5xx, errs[4] = addLastStat(lastData.Status5xx, newStats.Status5xx.Value, newStatsTime)
+ return lastData, util.JoinErrors(errs)
+}
+
+func addLastStatsToStatCacheStats(s dsdata.StatCacheStats, l dsdata.LastStatsData) dsdata.StatCacheStats {
+ s.Kbps.Value = l.Bytes.PerSec / BytesPerKilobit
+ s.Tps2xx.Value = l.Status2xx.PerSec
+ s.Tps3xx.Value = l.Status3xx.PerSec
+ s.Tps4xx.Value = l.Status4xx.PerSec
+ s.Tps5xx.Value = l.Status5xx.PerSec
+ s.TpsTotal.Value = s.Tps2xx.Value + s.Tps3xx.Value + s.Tps4xx.Value + s.Tps5xx.Value
+ return s
+}
+
+// addLastDSStatTotals takes a LastDSStat with only raw `Caches` data, and calculates and sets the `CacheGroups`, `Type`, and `Total` data, and returns the augmented structure.
+func addLastDSStatTotals(lastStat dsdata.LastDSStat, cachesReporting map[enum.CacheName]bool, serverCachegroups map[enum.CacheName]enum.CacheGroupName, serverTypes map[enum.CacheName]enum.CacheType) dsdata.LastDSStat {
+ cacheGroups := map[enum.CacheGroupName]dsdata.LastStatsData{}
+ cacheTypes := map[enum.CacheType]dsdata.LastStatsData{}
+ total := dsdata.LastStatsData{}
+ for cacheName, cacheStats := range lastStat.Caches {
+ if !cachesReporting[cacheName] {
+ continue
+ }
+
+ if cacheGroup, ok := serverCachegroups[cacheName]; ok {
+ cacheGroups[cacheGroup] = cacheGroups[cacheGroup].Sum(cacheStats)
+ } else {
+ log.Warnf("while computing delivery service data, cache %v not in cachegroups\n", cacheName)
+ }
+
+ if cacheType, ok := serverTypes[cacheName]; ok {
+ cacheTypes[cacheType] = cacheTypes[cacheType].Sum(cacheStats)
+ } else {
+ log.Warnf("while computing delivery service data, cache %v not in types\n", cacheName)
+ }
+ total = total.Sum(cacheStats)
+ }
+ lastStat.CacheGroups = cacheGroups
+ lastStat.Type = cacheTypes
+ lastStat.Total = total
+ return lastStat
+}
+
+// addDSPerSecStats calculates and adds the per-second delivery service stats to both the Stats and LastStats structures, and returns the augmented structures.
+func addDSPerSecStats(dsName enum.DeliveryServiceName, stat dsdata.Stat, lastStats dsdata.LastStats, dsStats dsdata.Stats, dsStatsTime time.Time, serverCachegroups map[enum.CacheName]enum.CacheGroupName, serverTypes map[enum.CacheName]enum.CacheType, mc to.TrafficMonitorConfigMap, events health.ThreadsafeEvents) (dsdata.Stats, dsdata.LastStats) {
+ err := error(nil)
+ lastStat, lastStatExists := lastStats.DeliveryServices[dsName]
+ if !lastStatExists {
+ lastStat = newLastDSStat()
+ }
+
+ for cacheName, cacheStats := range stat.Caches {
+ lastStat.Caches[cacheName], err = addLastStats(lastStat.Caches[cacheName], cacheStats, dsStatsTime)
+ if err != nil {
+ log.Warnf("%v adding kbps for cache %v: %v", dsName, cacheName, err)
+ continue
+ }
+ cacheStats.Kbps.Value = lastStat.Caches[cacheName].Bytes.PerSec / BytesPerKilobit
+ stat.Caches[cacheName] = cacheStats
+ }
+
+ lastStat = addLastDSStatTotals(lastStat, stat.CommonStats.CachesReporting, serverCachegroups, serverTypes)
+
+ for cacheGroup, cacheGroupStat := range lastStat.CacheGroups {
+ stat.CacheGroups[cacheGroup] = addLastStatsToStatCacheStats(stat.CacheGroups[cacheGroup], cacheGroupStat)
+ }
+ for cacheType, cacheTypeStat := range lastStat.Type {
+ stat.Types[cacheType] = addLastStatsToStatCacheStats(stat.Types[cacheType], cacheTypeStat)
+ }
+ stat.TotalStats = addLastStatsToStatCacheStats(stat.TotalStats, lastStat.Total)
+ lastStats.DeliveryServices[dsName] = lastStat
+ errStr := getDsErrString(dsName, stat.TotalStats, mc)
+ if errStr != "" {
+ stat.CommonStats.IsAvailable.Value = false
+ stat.CommonStats.IsHealthy.Value = false
+ stat.CommonStats.ErrorStr.Value = errStr
+ events.Add(health.Event{Time: time.Now(), Unix: time.Now().Unix(), Description: errStr, Name: dsName.String(), Hostname: dsName.String(), Type: "Delivery Service", Available: stat.CommonStats.IsAvailable.Value})
+ }
+
+ dsStats.DeliveryService[dsName] = stat
+ return dsStats, lastStats
+}
+
+// latestBytes returns the most recent OutBytes from the given cache results, and the time of that result. It assumes zero results are not valid, but nonzero results with errors are valid.
+func latestBytes(p cache.PrecomputedData) (int64, time.Time, error) {
+ if p.OutBytes == 0 {
+ return 0, time.Time{}, fmt.Errorf("no valid results")
+ }
+ return p.OutBytes, p.Time, nil
+}
+
+// addCachePerSecStats calculates the cache per-second stats, adds them to LastStats, and returns the augmented object.
+func addCachePerSecStats(cacheName enum.CacheName, precomputed cache.PrecomputedData, lastStats dsdata.LastStats) dsdata.LastStats {
+ outBytes, outBytesTime, err := latestBytes(precomputed) // it's ok if `latestBytes` returns 0s with an error, `addLastStat` will refrain from setting it (unless the previous calculation was nonzero, in which case it will error appropriately).
+ if err != nil {
+ log.Warnf("while computing delivery service data for cache %v: %v\n", cacheName, err)
+ }
+ lastStat := lastStats.Caches[cacheName] // if lastStats.Caches[cacheName] doesn't exist, it will be zero-constructed, and `addLastStat` will refrain from setting the PerSec for zero LastStats
+ lastStat.Bytes, err = addLastStat(lastStat.Bytes, outBytes, outBytesTime)
+ if err != nil {
+ log.Warnf("while computing delivery service data for cache %v: %v\n", cacheName, err)
+ return lastStats
+ }
+ lastStats.Caches[cacheName] = lastStat
+
+ return lastStats
+}
+
+// addPerSecStats adds Kbps fields to the NewStats, based on the previous out_bytes in the oldStats, and the time difference.
+//
+// Traffic Server only updates its data every N seconds. So, often we get a new Stats with the same OutBytes as the previous one,
+// So, we must record the last changed value, and the time it changed. Then, if the new OutBytes is different from the previous,
+// we set the (new - old) / lastChangedTime as the KBPS, and update the recorded LastChangedTime and LastChangedValue
+//
+// TODO handle ATS byte rolling (when the `out_bytes` overflows back to 0)
+func addPerSecStats(precomputed map[enum.CacheName]cache.PrecomputedData, dsStats dsdata.Stats, lastStats dsdata.LastStats, dsStatsTime time.Time, serverCachegroups map[enum.CacheName]enum.CacheGroupName, serverTypes map[enum.CacheName]enum.CacheType, mc to.TrafficMonitorConfigMap, events health.ThreadsafeEvents) (dsdata.Stats, dsdata.LastStats) {
+ for dsName, stat := range dsStats.DeliveryService {
+ dsStats, lastStats = addDSPerSecStats(dsName, stat, lastStats, dsStats, dsStatsTime, serverCachegroups, serverTypes, mc, events)
+ }
+ for cacheName, precomputedData := range precomputed {
+ lastStats = addCachePerSecStats(cacheName, precomputedData, lastStats)
+ }
+
+ return dsStats, lastStats
+}
+
+// CreateStats aggregates and creates statistics from given precomputed stat history. It returns the created stats, information about these stats necessary for the next calculation, and any error.
+func CreateStats(precomputed map[enum.CacheName]cache.PrecomputedData, toData todata.TOData, crStates peer.Crstates, lastStats dsdata.LastStats, now time.Time, mc to.TrafficMonitorConfigMap, events health.ThreadsafeEvents) (dsdata.Stats, dsdata.LastStats, error) {
+ start := time.Now()
+ dsStats := dsdata.NewStats()
+ for deliveryService := range toData.DeliveryServiceServers {
+ if deliveryService == "" {
+ log.Errorf("EMPTY CreateStats deliveryService")
+ continue
+ }
+ dsStats.DeliveryService[deliveryService] = *dsdata.NewStat()
+ }
+ dsStats = setStaticData(dsStats, toData.DeliveryServiceServers)
+ var err error
+ dsStats, err = addAvailableData(dsStats, crStates, toData.ServerCachegroups, toData.ServerDeliveryServices, toData.ServerTypes, precomputed) // TODO move after stat summarisation
+ if err != nil {
+ return dsStats, lastStats, fmt.Errorf("Error getting Cache availability data: %v", err)
+ }
+
+ for server, precomputedData := range precomputed {
+ cachegroup, ok := toData.ServerCachegroups[server]
+ if !ok {
+ log.Warnf("server %s has no cachegroup, skipping\n", server)
+ continue
+ }
+ serverType, ok := toData.ServerTypes[server]
+ if !ok {
+ log.Warnf("server %s not in CRConfig, skipping\n", server)
+ continue
+ }
+
+ // TODO check result.PrecomputedData.Errors
+ for ds, resultStat := range precomputedData.DeliveryServiceStats {
+ if ds == "" {
+ log.Errorf("EMPTY precomputed delivery service")
+ continue
+ }
+
+ if _, ok := dsStats.DeliveryService[ds]; !ok {
+ dsStats.DeliveryService[ds] = resultStat
+ continue
+ }
+ httpDsStat := dsStats.DeliveryService[ds]
+ httpDsStat.TotalStats = httpDsStat.TotalStats.Sum(resultStat.TotalStats)
+ httpDsStat.CacheGroups[cachegroup] = httpDsStat.CacheGroups[cachegroup].Sum(resultStat.CacheGroups[cachegroup])
+ httpDsStat.Types[serverType] = httpDsStat.Types[serverType].Sum(resultStat.Types[serverType])
+ httpDsStat.Caches[server] = httpDsStat.Caches[server].Sum(resultStat.Caches[server])
+ httpDsStat.CachesTimeReceived[server] = resultStat.CachesTimeReceived[server]
+ httpDsStat.CommonStats = dsStats.DeliveryService[ds].CommonStats
+ dsStats.DeliveryService[ds] = httpDsStat // TODO determine if necessary
+ }
+ }
+
+ perSecStats, lastStats := addPerSecStats(precomputed, dsStats, lastStats, now, toData.ServerCachegroups, toData.ServerTypes, mc, events)
+ log.Infof("CreateStats took %v\n", time.Since(start))
+ perSecStats.Time = time.Now()
+ return perSecStats, lastStats, nil
+}
+
+func getDsErrString(dsName enum.DeliveryServiceName, dsStats dsdata.StatCacheStats, monitorConfig to.TrafficMonitorConfigMap) string {
+ tpsThreshold := monitorConfig.DeliveryService[dsName.String()].TotalTPSThreshold
+ if tpsThreshold > 0 && dsStats.TpsTotal.Value > float64(tpsThreshold) {
+ return fmt.Sprintf("total.tps_total too high (%v > %v)", dsStats.TpsTotal.Value, tpsThreshold)
+ }
+
+ kbpsThreshold := monitorConfig.DeliveryService[dsName.String()].TotalKbpsThreshold
+ if kbpsThreshold > 0 && dsStats.Kbps.Value > float64(kbpsThreshold) {
+ return fmt.Sprintf("total.kbps too high (%v > %v)", dsStats.Kbps.Value, kbpsThreshold)
+ }
+ return ""
+}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/594b8517/traffic_monitor_golang/traffic_monitor/deliveryservicedata/stat.go
----------------------------------------------------------------------
diff --git a/traffic_monitor_golang/traffic_monitor/deliveryservicedata/stat.go b/traffic_monitor_golang/traffic_monitor/deliveryservicedata/stat.go
new file mode 100644
index 0000000..07d9f33
--- /dev/null
+++ b/traffic_monitor_golang/traffic_monitor/deliveryservicedata/stat.go
@@ -0,0 +1,488 @@
+package deliveryservicedata // TODO rename?
+
+/*
+ * 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 (
+ "errors"
+ "fmt"
+ "net/url"
+ "strconv"
+ "time"
+
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor_golang/traffic_monitor/enum"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor_golang/traffic_monitor/srvhttp"
+)
+
+// Filter encapsulates functions to filter a given set of Stats, e.g. from HTTP query parameters.
+// TODO combine with cache.Filter?
+type Filter interface {
+ UseStat(name string) bool
+ UseDeliveryService(name enum.DeliveryServiceName) bool
+ WithinStatHistoryMax(int) bool
+}
+
+// StatName is the name of a stat.
+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 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.
+type StatsOld struct {
+ DeliveryService map[enum.DeliveryServiceName]map[StatName][]StatOld `json:"deliveryService"`
+ srvhttp.CommonAPIData
+}
+
+// StatsReadonly is a read-only interface for delivery service Stats, designed to be passed to multiple goroutine readers.
+type StatsReadonly interface {
+ Get(enum.DeliveryServiceName) (StatReadonly, bool)
+ JSON(Filter, url.Values) StatsOld
+}
+
+// StatReadonly is a read-only interface for a delivery service Stat, designed to be passed to multiple goroutine readers.
+type StatReadonly interface {
+ Copy() Stat
+ Common() StatCommonReadonly
+ CacheGroup(name enum.CacheGroupName) (StatCacheStats, bool)
+ Type(name enum.CacheType) (StatCacheStats, bool)
+ Total() StatCacheStats
+}
+
+// StatCommonReadonly is a read-only interface for a delivery service's common Stat data, designed to be passed to multiple goroutine readers.
+type StatCommonReadonly interface {
+ Copy() StatCommon
+ CachesConfigured() StatInt
+ CachesReportingNames() []enum.CacheName
+ Error() StatString
+ Status() StatString
+ Healthy() StatBool
+ Available() StatBool
+ CachesAvailable() StatInt
+}
+
+// StatMeta includes metadata about a particular stat.
+type StatMeta struct {
+ Time int64 `json:"time"`
+}
+
+// StatFloat is a float stat, combined with its metadata
+type StatFloat struct {
+ StatMeta
+ Value float64 `json:"value"`
+}
+
+// StatBool is a boolean stat, combined with its metadata
+type StatBool struct {
+ StatMeta
+ Value bool `json:"value"`
+}
+
+// StatInt is an integer stat, combined with its metadata
+type StatInt struct {
+ StatMeta
+ Value int64 `json:"value"`
+}
+
+// StatString is a string stat, combined with its metadata
+type StatString struct {
+ StatMeta
+ Value string `json:"value"`
+}
+
+// StatCommon contains stat data common to most delivery service stats.
+type StatCommon struct {
+ CachesConfiguredNum StatInt `json:"caches_configured"`
+ CachesReporting map[enum.CacheName]bool `json:"caches_reporting"`
+ ErrorStr StatString `json:"error_string"`
+ StatusStr StatString `json:"status"`
+ 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.
+func (a StatCommon) Copy() StatCommon {
+ b := a
+ 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
+}
+
+// CachesConfigured returns the number of caches configured for this delivery service stat. It is part of the StatCommonReadonly interface.
+func (a StatCommon) CachesConfigured() StatInt {
+ return a.CachesConfiguredNum
+}
+
+// CacheReporting returns the number of caches reporting for this delivery service stat. It is part of the StatCommonReadonly interface.
+func (a StatCommon) CacheReporting(name enum.CacheName) (bool, bool) {
+ c, ok := a.CachesReporting[name]
+ return c, ok
+}
+
+// CachesReportingNames returns the list of caches reporting for this delivery service stat. It is part of the StatCommonReadonly interface.
+func (a StatCommon) CachesReportingNames() []enum.CacheName {
+ names := make([]enum.CacheName, 0, len(a.CachesReporting))
+ for name := range a.CachesReporting {
+ names = append(names, name)
+ }
+ return names
+}
+
+// Error returns the error string of this delivery service stat. It is part of the StatCommonReadonly interface.
+func (a StatCommon) Error() StatString {
+ return a.ErrorStr
+}
+
+// Status returns the status string of this delivery service stat. It is part of the StatCommonReadonly interface.
+func (a StatCommon) Status() StatString {
+ return a.StatusStr
+}
+
+// Healthy returns whether this delivery service is considered healthy by this stat. It is part of the StatCommonReadonly interface.
+func (a StatCommon) Healthy() StatBool {
+ return a.IsHealthy
+}
+
+// Available returns whether this delivery service is considered available by this stat. It is part of the StatCommonReadonly interface.
+func (a StatCommon) Available() StatBool {
+ return a.IsAvailable
+}
+
+// CachesAvailable returns the number of caches available to the delivery service in this stat. It is part of the StatCommonReadonly interface.
+func (a StatCommon) CachesAvailable() StatInt {
+ return a.CachesAvailableNum
+}
+
+// StatCacheStats is all the stats generated by a cache.
+// This may also be used for aggregate stats, for example, the summary of all cache stats for a cache group, or delivery service.
+// Each stat is an array, in case there are multiple data points at different times. However, a single data point i.e. a single array member is common.
+type StatCacheStats struct {
+ OutBytes StatInt `json:"out_bytes"`
+ IsAvailable StatBool `json:"is_available"`
+ Status5xx StatInt `json:"status_5xx"`
+ Status4xx StatInt `json:"status_4xx"`
+ Status3xx StatInt `json:"status_3xx"`
+ Status2xx StatInt `json:"status_2xx"`
+ InBytes StatFloat `json:"in_bytes"`
+ Kbps StatFloat `json:"kbps"`
+ Tps5xx StatFloat `json:"tps_5xx"`
+ Tps4xx StatFloat `json:"tps_4xx"`
+ Tps3xx StatFloat `json:"tps_3xx"`
+ Tps2xx StatFloat `json:"tps_2xx"`
+ ErrorString StatString `json:"error_string"`
+ TpsTotal StatFloat `json:"tps_total"`
+}
+
+// Sum adds the given cache stats to this cache stats. Numeric values are summed; strings are appended.
+func (a StatCacheStats) Sum(b StatCacheStats) StatCacheStats {
+ return StatCacheStats{
+ OutBytes: StatInt{Value: a.OutBytes.Value + b.OutBytes.Value},
+ IsAvailable: StatBool{Value: a.IsAvailable.Value || b.IsAvailable.Value},
+ Status5xx: StatInt{Value: a.Status5xx.Value + b.Status5xx.Value},
+ Status4xx: StatInt{Value: a.Status4xx.Value + b.Status4xx.Value},
+ Status3xx: StatInt{Value: a.Status3xx.Value + b.Status3xx.Value},
+ Status2xx: StatInt{Value: a.Status2xx.Value + b.Status2xx.Value},
+ InBytes: StatFloat{Value: a.InBytes.Value + b.InBytes.Value},
+ Kbps: StatFloat{Value: a.Kbps.Value + b.Kbps.Value},
+ Tps5xx: StatFloat{Value: a.Tps5xx.Value + b.Tps5xx.Value},
+ Tps4xx: StatFloat{Value: a.Tps4xx.Value + b.Tps4xx.Value},
+ Tps3xx: StatFloat{Value: a.Tps3xx.Value + b.Tps3xx.Value},
+ Tps2xx: StatFloat{Value: a.Tps2xx.Value + b.Tps2xx.Value},
+ ErrorString: StatString{Value: a.ErrorString.Value + b.ErrorString.Value},
+ TpsTotal: StatFloat{Value: a.TpsTotal.Value + b.TpsTotal.Value},
+ }
+}
+
+// Stat represents a complete delivery service stat, for a given poll, or at the time requested.
+type Stat struct {
+ CommonStats StatCommon
+ CacheGroups map[enum.CacheGroupName]StatCacheStats
+ Types map[enum.CacheType]StatCacheStats
+ Caches map[enum.CacheName]StatCacheStats
+ CachesTimeReceived map[enum.CacheName]time.Time
+ TotalStats StatCacheStats
+}
+
+// ErrNotProcessedStat indicates a stat received is not used by Traffic Monitor, nor returned by any API endpoint. Receiving this error indicates the stat has been discarded.
+var ErrNotProcessedStat = errors.New("This stat is not used.")
+
+// NewStat returns a new delivery service Stat, initializing pointer members.
+func NewStat() *Stat {
+ return &Stat{
+ CacheGroups: map[enum.CacheGroupName]StatCacheStats{},
+ Types: map[enum.CacheType]StatCacheStats{},
+ CommonStats: StatCommon{CachesReporting: map[enum.CacheName]bool{}},
+ Caches: map[enum.CacheName]StatCacheStats{},
+ CachesTimeReceived: map[enum.CacheName]time.Time{},
+ }
+}
+
+// Copy performs a deep copy of this Stat. It does not modify, and is thus safe for multiple goroutines.
+func (a Stat) Copy() Stat {
+ b := Stat{
+ CommonStats: a.CommonStats.Copy(),
+ TotalStats: a.TotalStats,
+ CacheGroups: map[enum.CacheGroupName]StatCacheStats{},
+ Types: map[enum.CacheType]StatCacheStats{},
+ Caches: map[enum.CacheName]StatCacheStats{},
+ CachesTimeReceived: map[enum.CacheName]time.Time{},
+ }
+ for k, v := range a.CacheGroups {
+ b.CacheGroups[k] = v
+ }
+ for k, v := range a.Types {
+ b.Types[k] = v
+ }
+ for k, v := range a.Caches {
+ b.Caches[k] = v
+ }
+ for k, v := range a.CachesTimeReceived {
+ b.CachesTimeReceived[k] = v
+ }
+ return b
+}
+
+// Common returns the common stat data for this stat. It is part of the StatCommonReadonly interface.
+func (a Stat) Common() StatCommonReadonly {
+ return a.CommonStats
+}
+
+// CacheGroup returns the data for the given cachegroup in this stat. It is part of the StatCommonReadonly interface.
+func (a Stat) CacheGroup(name enum.CacheGroupName) (StatCacheStats, bool) {
+ c, ok := a.CacheGroups[name]
+ return c, ok
+}
+
+// Type returns the aggregated data for the given cache type in this stat. It is part of the StatCommonReadonly interface.
+func (a Stat) Type(name enum.CacheType) (StatCacheStats, bool) {
+ t, ok := a.Types[name]
+ return t, ok
+}
+
+// Total returns the aggregated total data in this stat. It is part of the StatCommonReadonly interface.
+func (a Stat) Total() StatCacheStats {
+ return a.TotalStats
+}
+
+// Stats is the JSON-serialisable representation of delivery service Stats. It maps delivery service names to individual stat objects.
+type Stats struct {
+ DeliveryService map[enum.DeliveryServiceName]Stat `json:"deliveryService"`
+ Time time.Time `json:"-"`
+}
+
+// Copy performs a deep copy of this Stats object.
+func (s Stats) Copy() Stats {
+ b := NewStats()
+ for k, v := range s.DeliveryService {
+ b.DeliveryService[k] = v.Copy()
+ }
+ b.Time = s.Time
+ return b
+}
+
+// Get returns the stats for the given delivery service, and whether it exists.
+func (s Stats) Get(name enum.DeliveryServiceName) (StatReadonly, bool) {
+ ds, ok := s.DeliveryService[name]
+ return ds, ok
+}
+
+// JSON returns an object formatted as expected to be serialized to JSON and served.
+func (s Stats) JSON(filter Filter, params url.Values) StatsOld {
+ // TODO fix to be the time calculated, not the time requested
+ now := s.Time.UnixNano() / int64(time.Millisecond) // Traffic Monitor 1.0 API is 'ms since the epoch'
+ jsonObj := &StatsOld{
+ CommonAPIData: srvhttp.GetCommonAPIData(params, time.Now()),
+ DeliveryService: map[enum.DeliveryServiceName]map[StatName][]StatOld{},
+ }
+
+ for deliveryService, stat := range s.DeliveryService {
+ if !filter.UseDeliveryService(deliveryService) {
+ continue
+ }
+ jsonObj.DeliveryService[deliveryService] = map[StatName][]StatOld{}
+ jsonObj = addCommonData(jsonObj, &stat.CommonStats, deliveryService, now, filter)
+ for cacheGroup, cacheGroupStats := range stat.CacheGroups {
+ jsonObj = addStatCacheStats(jsonObj, cacheGroupStats, deliveryService, "location."+string(cacheGroup)+".", now, filter)
+ }
+ for cacheType, typeStats := range stat.Types {
+ jsonObj = addStatCacheStats(jsonObj, typeStats, deliveryService, "type."+cacheType.String()+".", now, filter)
+ }
+ jsonObj = addStatCacheStats(jsonObj, stat.TotalStats, deliveryService, "total.", now, filter)
+ }
+ return *jsonObj
+}
+
+// NewStats creates a new Stats object, initializing any pointer members.
+// TODO rename to just 'New'?
+func NewStats() Stats {
+ return Stats{DeliveryService: map[enum.DeliveryServiceName]Stat{}}
+}
+
+// LastStats includes the previously recieved stats for DeliveryServices and Caches, the stat itself, when it was received, and the stat value per second.
+type LastStats struct {
+ DeliveryServices map[enum.DeliveryServiceName]LastDSStat
+ Caches map[enum.CacheName]LastStatsData
+}
+
+// NewLastStats returns a new LastStats object, initializing internal pointer values.
+func NewLastStats() LastStats {
+ return LastStats{DeliveryServices: map[enum.DeliveryServiceName]LastDSStat{}, Caches: map[enum.CacheName]LastStatsData{}}
+}
+
+// Copy performs a deep copy of this LastStats object.
+func (a LastStats) Copy() LastStats {
+ b := NewLastStats()
+ for k, v := range a.DeliveryServices {
+ b.DeliveryServices[k] = v.Copy()
+ }
+ for k, v := range a.Caches {
+ b.Caches[k] = v
+ }
+ return b
+}
+
+// LastDSStat maps and aggregates the last stats received for the given delivery service to caches, cache groups, types, and total.
+// TODO figure a way to associate this type with StatHTTP, with which its members correspond.
+type LastDSStat struct {
+ Caches map[enum.CacheName]LastStatsData
+ CacheGroups map[enum.CacheGroupName]LastStatsData
+ Type map[enum.CacheType]LastStatsData
+ Total LastStatsData
+}
+
+// Copy performs a deep copy of this LastDSStat object.
+func (a LastDSStat) Copy() LastDSStat {
+ b := LastDSStat{
+ CacheGroups: map[enum.CacheGroupName]LastStatsData{},
+ Type: map[enum.CacheType]LastStatsData{},
+ Caches: map[enum.CacheName]LastStatsData{},
+ Total: a.Total,
+ }
+ for k, v := range a.CacheGroups {
+ b.CacheGroups[k] = v
+ }
+ for k, v := range a.Type {
+ b.Type[k] = v
+ }
+ for k, v := range a.Caches {
+ b.Caches[k] = v
+ }
+ return b
+}
+
+func newLastDSStat() LastDSStat {
+ return LastDSStat{
+ CacheGroups: map[enum.CacheGroupName]LastStatsData{},
+ Type: map[enum.CacheType]LastStatsData{},
+ Caches: map[enum.CacheName]LastStatsData{},
+ }
+}
+
+// LastStatsData contains the last stats and per-second calculations for bytes and status codes received from a cache.
+type LastStatsData struct {
+ Bytes LastStatData
+ Status2xx LastStatData
+ Status3xx LastStatData
+ Status4xx LastStatData
+ Status5xx LastStatData
+}
+
+// Sum returns the Sum() of each member data with the given LastStatsData corresponding members
+func (a LastStatsData) Sum(b LastStatsData) LastStatsData {
+ return LastStatsData{
+ Bytes: a.Bytes.Sum(b.Bytes),
+ Status2xx: a.Status2xx.Sum(b.Status2xx),
+ Status3xx: a.Status3xx.Sum(b.Status3xx),
+ Status4xx: a.Status4xx.Sum(b.Status4xx),
+ Status5xx: a.Status5xx.Sum(b.Status5xx),
+ }
+}
+
+// LastStatData contains the value, time it was received, and per-second calculation since the previous stat, for a stat from a cache.
+type LastStatData struct {
+ PerSec float64
+ Stat int64
+ Time time.Time
+}
+
+// Sum adds the PerSec and Stat of the given data to this object. Time is meaningless for the summed object, and is thus set to 0.
+func (a LastStatData) Sum(b LastStatData) LastStatData {
+ return LastStatData{
+ PerSec: a.PerSec + b.PerSec,
+ Stat: a.Stat + b.Stat,
+ }
+}
+
+func addCommonData(s *StatsOld, c *StatCommon, deliveryService enum.DeliveryServiceName, t int64, filter Filter) *StatsOld {
+ add := func(name string, val interface{}) {
+ if filter.UseStat(name) {
+ s.DeliveryService[deliveryService][StatName(name)] = []StatOld{StatOld{Time: t, Value: val}}
+ }
+ }
+ 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", fmt.Sprintf("%d", c.CachesAvailableNum.Value))
+ add("disabledLocations", c.CachesDisabled)
+ return s
+}
+
+func addStatCacheStats(s *StatsOld, c StatCacheStats, deliveryService enum.DeliveryServiceName, prefix string, t int64, filter Filter) *StatsOld {
+ add := func(name, val string) {
+ if filter.UseStat(name) {
+ // This is for compatibility with the Traffic Monitor 1.0 API.
+ // TODO abstract this? Or deprecate and remove it?
+ if name == "isAvailable" || name == "error-string" {
+ s.DeliveryService[deliveryService][StatName("location."+prefix+name)] = []StatOld{StatOld{Time: t, Value: val}}
+ } else {
+ s.DeliveryService[deliveryService][StatName(prefix+name)] = []StatOld{StatOld{Time: t, Value: val}}
+ }
+ }
+ }
+ add("out_bytes", strconv.Itoa(int(c.OutBytes.Value)))
+ add("isAvailable", fmt.Sprintf("%t", c.IsAvailable.Value))
+ add("status_5xx", strconv.Itoa(int(c.Status5xx.Value)))
+ add("status_4xx", strconv.Itoa(int(c.Status4xx.Value)))
+ add("status_3xx", strconv.Itoa(int(c.Status3xx.Value)))
+ add("status_2xx", strconv.Itoa(int(c.Status2xx.Value)))
+ add("in_bytes", strconv.Itoa(int(c.InBytes.Value)))
+ add("kbps", strconv.Itoa(int(c.Kbps.Value)))
+ add("tps_5xx", fmt.Sprintf("%f", c.Tps5xx.Value))
+ add("tps_4xx", fmt.Sprintf("%f", c.Tps4xx.Value))
+ add("tps_3xx", fmt.Sprintf("%f", c.Tps3xx.Value))
+ add("tps_2xx", fmt.Sprintf("%f", c.Tps2xx.Value))
+ add("error-string", c.ErrorString.Value)
+ add("tps_total", fmt.Sprintf("%f", c.TpsTotal.Value))
+ return s
+}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/594b8517/traffic_monitor_golang/traffic_monitor/enum/enum.go
----------------------------------------------------------------------
diff --git a/traffic_monitor_golang/traffic_monitor/enum/enum.go b/traffic_monitor_golang/traffic_monitor/enum/enum.go
new file mode 100644
index 0000000..45546bd
--- /dev/null
+++ b/traffic_monitor_golang/traffic_monitor/enum/enum.go
@@ -0,0 +1,181 @@
+// Package enum contains enumerations and strongly typed names.
+//
+// These enums should be treated as enumerables, and MUST NOT be cast as anything else (integer, strings, etc). Enums MUST NOT be compared to strings or integers via casting. Enumerable data SHOULD be stored as the enumeration, not as a string or number. The *only* reason they are internally represented as strings, is to make them implicitly serialize to human-readable JSON. They should not be treated as strings. Casting or storing strings or numbers defeats the purpose of enum safety and conveniences.
+//
+// When storing enumumerable data in memory, it SHOULD be converted to and stored as an enum via the corresponding `FromString` function, checked whether the conversion failed and Invalid values handled, and valid data stored as the enum. This guarantees stored data is valid, and catches invalid input as soon as possible.
+//
+// When adding new enum types, enums should be internally stored as strings, so they implicitly serialize as human-readable JSON, unless the performance or memory of integers is necessary (it almost certainly isn't). Enums should always have the "invalid" value as the empty string (or 0), so default-initialized enums are invalid.
+// Enums should always have a FromString() conversion function, to convert input data to enums. Conversion functions should usually be case-insensitive, and may ignore underscores or hyphens, depending on the use case.
+//
+package enum
+
+/*
+ * 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 (
+ "strings"
+)
+
+// TrafficMonitorName is the hostname of a Traffic Monitor peer.
+type TrafficMonitorName string
+
+// CacheName is the hostname of a CDN cache.
+type CacheName string
+
+// CacheGroupName is the name of a CDN cachegroup.
+type CacheGroupName string
+
+// DeliveryServiceName is the name of a CDN delivery service.
+type DeliveryServiceName string
+
+// CacheType is the type (or tier) of a CDN cache.
+type CacheType string
+
+const (
+ // CacheTypeEdge represents an edge cache.
+ CacheTypeEdge = CacheType("EDGE")
+ // CacheTypeMid represents a mid cache.
+ CacheTypeMid = CacheType("MID")
+ // CacheTypeInvalid represents an cache type enumeration. Note this is the default construction for a CacheType.
+ CacheTypeInvalid = CacheType("")
+)
+
+func (c CacheName) String() string {
+ return string(c)
+}
+
+func (t TrafficMonitorName) String() string {
+ return string(t)
+}
+
+func (d DeliveryServiceName) String() string {
+ return string(d)
+}
+
+// String returns a string representation of this cache type.
+func (t CacheType) String() string {
+ switch t {
+ case CacheTypeEdge:
+ return "EDGE"
+ case CacheTypeMid:
+ return "MID"
+ default:
+ return "INVALIDCACHETYPE"
+ }
+}
+
+// CacheTypeFromString returns a cache type object from its string representation, or CacheTypeInvalid if the string is not a valid type.
+func CacheTypeFromString(s string) CacheType {
+ s = strings.ToLower(s)
+ if strings.HasPrefix(s, "edge") {
+ return CacheTypeEdge
+ }
+ if strings.HasPrefix(s, "mid") {
+ return CacheTypeMid
+ }
+ return CacheTypeInvalid
+}
+
+// DSType is the Delivery Service type. HTTP, DNS, etc.
+type DSType string
+
+const (
+ // DSTypeHTTP represents an HTTP delivery service
+ DSTypeHTTP = DSType("http")
+ // DSTypeDNS represents a DNS delivery service
+ DSTypeDNS = DSType("dns")
+ // DSTypeInvalid represents an invalid delivery service type enumeration. Note this is the default construction for a DSType.
+ DSTypeInvalid = DSType("")
+)
+
+// String returns a string representation of this delivery service type.
+func (t DSType) String() string {
+ switch t {
+ case DSTypeHTTP:
+ return "HTTP"
+ case DSTypeDNS:
+ return "DNS"
+ default:
+ return "INVALIDDSTYPE"
+ }
+}
+
+// DSTypeFromString returns a delivery service type object from its string representation, or DSTypeInvalid if the string is not a valid type.
+func DSTypeFromString(s string) DSType {
+ s = strings.ToLower(s)
+ switch s {
+ case "http":
+ return DSTypeHTTP
+ case "dns":
+ return DSTypeDNS
+ default:
+ return DSTypeInvalid
+ }
+}
+
+// CacheStatus represents the Traffic Server status set in Traffic Ops (online, offline, admin_down, reported). The string values of this type should match the Traffic Ops values.
+type CacheStatus string
+
+const (
+ // CacheStatusAdminDown represents a cache which has been administratively marked as down, but which should still appear in the CDN (Traffic Server, Traffic Monitor, Traffic Router).
+ CacheStatusAdminDown = CacheStatus("ADMIN_DOWN")
+ // CacheStatusOnline represents a cache which has been marked as Online in Traffic Ops, irrespective of monitoring. Traffic Monitor will always flag these caches as available.
+ CacheStatusOnline = CacheStatus("ONLINE")
+ // CacheStatusOffline represents a cache which has been marked as Offline in Traffic Ops. These caches will not be returned in any endpoint, and Traffic Monitor acts like they don't exist.
+ CacheStatusOffline = CacheStatus("OFFLINE")
+ // CacheStatusReported represents a cache which has been marked as Reported in Traffic Ops. These caches are polled for health and returned in endpoints as available or unavailable based on bandwidth, response time, and other factors. The vast majority of caches should be Reported.
+ CacheStatusReported = CacheStatus("REPORTED")
+ // CacheStatusInvalid represents an invalid status enumeration.
+ CacheStatusInvalid = CacheStatus("")
+)
+
+// String returns a string representation of this cache status
+func (t CacheStatus) String() string {
+ switch t {
+ case CacheStatusAdminDown:
+ fallthrough
+ case CacheStatusOnline:
+ fallthrough
+ case CacheStatusOffline:
+ fallthrough
+ case CacheStatusReported:
+ return string(t)
+ default:
+ return "INVALIDCACHESTATUS"
+ }
+}
+
+// CacheStatusFromString returns a CacheStatus from its string representation, or CacheStatusInvalid if the string is not a valid type.
+func CacheStatusFromString(s string) CacheStatus {
+ s = strings.ToLower(s)
+ switch s {
+ case "admin_down":
+ fallthrough
+ case "admindown":
+ return CacheStatusAdminDown
+ case "offline":
+ return CacheStatusOffline
+ case "online":
+ return CacheStatusOnline
+ case "reported":
+ return CacheStatusReported
+ default:
+ return CacheStatusInvalid
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/594b8517/traffic_monitor_golang/traffic_monitor/health/cache.go
----------------------------------------------------------------------
diff --git a/traffic_monitor_golang/traffic_monitor/health/cache.go b/traffic_monitor_golang/traffic_monitor/health/cache.go
new file mode 100644
index 0000000..7cf6eb5
--- /dev/null
+++ b/traffic_monitor_golang/traffic_monitor/health/cache.go
@@ -0,0 +1,272 @@
+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"
+ "time"
+
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor_golang/common/log"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor_golang/common/util"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor_golang/traffic_monitor/cache"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor_golang/traffic_monitor/enum"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor_golang/traffic_monitor/peer"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor_golang/traffic_monitor/threadsafe"
+ todata "github.com/apache/incubator-trafficcontrol/traffic_monitor_golang/traffic_monitor/trafficopsdata"
+ to "github.com/apache/incubator-trafficcontrol/traffic_ops/client"
+)
+
+// 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 {
+ setErr(newResult, fmt.Errorf("Error converting load average string '%s': %v", newResult.Astats.System.ProcLoadavg, err))
+ return
+ }
+ newResult.Vitals.LoadAvg = oneMinAvg
+ } else {
+ setErr(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 {
+ 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 {
+ setErr(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 {
+ setErr(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, eventDesc(status, availability+"; invalid status"), ""
+ case status == enum.CacheStatusAdminDown:
+ return false, eventDesc(status, availability), ""
+ case status == enum.CacheStatusOffline:
+ log.Errorf("Cache %v set to offline, but still polled\n", result.ID)
+ return false, eventDesc(status, availability), ""
+ case status == enum.CacheStatusOnline:
+ return true, eventDesc(status, availability), ""
+ case result.Error != nil:
+ return false, eventDesc(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 := util.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, 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 {
+ switch threshold.Comparator {
+ case "=":
+ return fmt.Sprintf("%s not equal (%.2f != %.2f)", stat, val, threshold.Val)
+ case ">":
+ return fmt.Sprintf("%s too low (%.2f < %.2f)", stat, val, threshold.Val)
+ case "<":
+ return fmt.Sprintf("%s too high (%.2f > %.2f)", stat, val, threshold.Val)
+ case ">=":
+ return fmt.Sprintf("%s too low (%.2f <= %.2f)", stat, val, threshold.Val)
+ case "<=":
+ return fmt.Sprintf("%s too high (%.2f >= %.2f)", 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 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/594b8517/traffic_monitor_golang/traffic_monitor/health/event.go
----------------------------------------------------------------------
diff --git a/traffic_monitor_golang/traffic_monitor/health/event.go b/traffic_monitor_golang/traffic_monitor/health/event.go
new file mode 100644
index 0000000..4dd40a3
--- /dev/null
+++ b/traffic_monitor_golang/traffic_monitor/health/event.go
@@ -0,0 +1,83 @@
+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 (
+ "sync"
+ "time"
+
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor_golang/common/log"
+)
+
+// Event represents an event change in aggregated data. For example, a cache being marked as unavailable.
+type Event struct {
+ Time time.Time `json:"-"`
+ Index uint64 `json:"index"`
+ Unix int64 `json:"time"`
+ Description string `json:"description"`
+ Name string `json:"name"`
+ Hostname string `json:"hostname"`
+ 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()
+}