You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@trafficcontrol.apache.org by mi...@apache.org on 2016/11/07 22:23:41 UTC
[20/26] incubator-trafficcontrol git commit: merge master
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/be5ab888/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 a2cc876..023c570 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/healthresult.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/healthresult.go
@@ -1,25 +1,48 @@
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/Comcast/traffic_control/traffic_monitor/experimental/common/log"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/cache"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/config"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/enum"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/health"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/peer"
- todata "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/trafficopsdata"
+ "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"
+ 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 {
@@ -28,6 +51,7 @@ func (a DurationMap) Copy() DurationMap {
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}
@@ -40,6 +64,7 @@ func (o *DurationMapThreadsafe) Get() DurationMap {
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
@@ -84,9 +109,6 @@ func StartHealthResultManager(
return lastHealthDurations, events, localCacheStatus
}
-// cacheAggregateSeconds is how often to aggregate stats, if the health chan is never empty. (Otherwise, we read from the chan until it's empty, then aggregate, continuously)
-const cacheAggregateSeconds = 1
-
func healthResultManagerListen(
cacheHealthChan <-chan cache.Result,
toData todata.TODataThreadsafe,
@@ -192,7 +214,7 @@ func processHealthResult(
log.Debugf("poll %v %v healthresultman start\n", healthResult.PollID, time.Now())
fetchCount.Inc()
var prevResult cache.Result
- healthResultHistory := healthHistory[enum.CacheName(healthResult.Id)]
+ healthResultHistory := healthHistory[healthResult.ID]
if len(healthResultHistory) != 0 {
prevResult = healthResultHistory[len(healthResultHistory)-1]
}
@@ -201,16 +223,17 @@ func processHealthResult(
health.GetVitals(&healthResult, &prevResult, &monitorConfigCopy)
}
- healthHistory[enum.CacheName(healthResult.Id)] = pruneHistory(append(healthHistory[enum.CacheName(healthResult.Id)], healthResult), cfg.MaxHealthHistory)
+ maxHistory := uint64(monitorConfigCopy.Profile[monitorConfigCopy.TrafficServer[string(healthResult.ID)].Profile].Parameters.HistoryCount)
+ healthHistory[healthResult.ID] = pruneHistory(append(healthHistory[healthResult.ID], healthResult), maxHistory)
isAvailable, whyAvailable := health.EvalCache(healthResult, &monitorConfigCopy)
- if localStates.Get().Caches[healthResult.Id].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(Event{Time: time.Now().Unix(), Description: whyAvailable, Name: healthResult.Id, Hostname: healthResult.Id, Type: toDataCopy.ServerTypes[healthResult.Id].String(), Available: isAvailable})
+ if localStates.Get().Caches[healthResult.ID].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(Event{Time: time.Now().Unix(), Description: whyAvailable, Name: healthResult.ID, Hostname: healthResult.ID, Type: toDataCopy.ServerTypes[healthResult.ID].String(), Available: isAvailable})
}
- localCacheStatus[healthResult.Id] = CacheAvailableStatus{Available: isAvailable, Status: monitorConfigCopy.TrafficServer[string(healthResult.Id)].Status} // TODO move within localStates?
- localStates.SetCache(healthResult.Id, peer.IsAvailable{IsAvailable: isAvailable})
+ localCacheStatus[healthResult.ID] = CacheAvailableStatus{Available: isAvailable, Status: monitorConfigCopy.TrafficServer[string(healthResult.ID)].Status} // TODO move within localStates?
+ localStates.SetCache(healthResult.ID, peer.IsAvailable{IsAvailable: isAvailable})
log.Debugf("poll %v %v calculateDeliveryServiceState start\n", healthResult.PollID, time.Now())
calculateDeliveryServiceState(toDataCopy.DeliveryServiceServers, localStates)
log.Debugf("poll %v %v calculateDeliveryServiceState end\n", healthResult.PollID, time.Now())
@@ -220,11 +243,11 @@ func processHealthResult(
lastHealthDurations := lastHealthDurationsThreadsafe.Get().Copy()
for _, healthResult := range results {
- if lastHealthStart, ok := lastHealthEndTimes[enum.CacheName(healthResult.Id)]; ok {
+ if lastHealthStart, ok := lastHealthEndTimes[healthResult.ID]; ok {
d := time.Since(lastHealthStart)
- lastHealthDurations[enum.CacheName(healthResult.Id)] = d
+ lastHealthDurations[healthResult.ID] = d
}
- lastHealthEndTimes[enum.CacheName(healthResult.Id)] = time.Now()
+ lastHealthEndTimes[healthResult.ID] = time.Now()
log.Debugf("poll %v %v finish\n", healthResult.PollID, time.Now())
healthResult.PollFinished <- healthResult.PollID
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/be5ab888/traffic_monitor/experimental/traffic_monitor/manager/lastkbpsstats.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/lastkbpsstats.go b/traffic_monitor/experimental/traffic_monitor/manager/lastkbpsstats.go
index 103ebd0..cb5811d 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/lastkbpsstats.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/lastkbpsstats.go
@@ -1,15 +1,37 @@
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 (
- ds "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/deliveryservice"
+ ds "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/deliveryservice"
"sync"
)
+// LastStatsThreadsafe wraps a deliveryservice.LastStats object to be safe for multiple readers and one writer.
type LastStatsThreadsafe struct {
stats *ds.LastStats
m *sync.RWMutex
}
+// NewLastStatsThreadsafe returns a wrapped a deliveryservice.LastStats object safe for multiple readers and one writer.
func NewLastStatsThreadsafe() LastStatsThreadsafe {
s := ds.NewLastStats()
return LastStatsThreadsafe{m: &sync.RWMutex{}, stats: &s}
@@ -22,6 +44,7 @@ func (o *LastStatsThreadsafe) Get() ds.LastStats {
return *o.stats
}
+// Set sets the internal LastStats object. This MUST NOT be called by multiple goroutines.
func (o *LastStatsThreadsafe) Set(s ds.LastStats) {
o.m.Lock()
*o.stats = s
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/be5ab888/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 d86fad1..49fa13a 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/manager.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/manager.go
@@ -1,22 +1,43 @@
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 (
"crypto/tls"
"net/http"
"time"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/common/fetcher"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/common/handler"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/common/poller"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/cache"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/config"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/peer"
- todata "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/trafficopsdata"
- towrap "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/trafficopswrapper"
- // to "github.com/Comcast/traffic_control/traffic_ops/client"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/common/fetcher"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/common/handler"
+ "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/peer"
+ todata "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/trafficopsdata"
+ towrap "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/trafficopswrapper"
+ // to "github.com/apache/incubator-trafficcontrol/traffic_ops/client"
"github.com/davecheney/gmx"
)
+// StaticAppData encapsulates data about the app available at startup
type StaticAppData struct {
StartTime time.Time
GitRevision string
@@ -29,7 +50,7 @@ type StaticAppData struct {
}
//
-// Kicks off the pollers and handlers
+// Start starts the poller and handler goroutines
//
func Start(opsConfigFile string, cfg config.Config, staticAppData StaticAppData) {
toSession := towrap.ITrafficOpsSession(towrap.NewTrafficOpsSessionThreadsafe(nil))
@@ -41,7 +62,6 @@ func Start(opsConfigFile string, cfg config.Config, staticAppData StaticAppData)
// TODO investigate whether a unique client per cache to be polled is faster
sharedClient := &http.Client{
- Timeout: cfg.HttpTimeout,
Transport: &http.Transport{TLSClientConfig: &tls.Config{InsecureSkipVerify: true}},
}
@@ -130,6 +150,8 @@ func Start(opsConfigFile string, cfg config.Config, staticAppData StaticAppData)
errorCount,
localCacheStatus,
unpolledCaches,
+ monitorConfig,
+ cfg,
)
healthTickListener(cacheHealthPoller.TickChan, healthIteration)
@@ -137,10 +159,7 @@ func Start(opsConfigFile string, cfg config.Config, staticAppData StaticAppData)
// healthTickListener listens for health ticks, and writes to the health iteration variable. Does not return.
func healthTickListener(cacheHealthTick <-chan uint64, healthIteration UintThreadsafe) {
- for {
- select {
- case i := <-cacheHealthTick:
- healthIteration.Set(i)
- }
+ for i := range cacheHealthTick {
+ healthIteration.Set(i)
}
}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/be5ab888/traffic_monitor/experimental/traffic_monitor/manager/monitorconfig.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/monitorconfig.go b/traffic_monitor/experimental/traffic_monitor/manager/monitorconfig.go
index fe42dad..5c343ae 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/monitorconfig.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/monitorconfig.go
@@ -1,17 +1,40 @@
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 (
"fmt"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/common/log"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/common/poller"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/config"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/enum"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/peer"
- to "github.com/Comcast/traffic_control/traffic_ops/client"
"strings"
"sync"
+ "time"
+
+ "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/enum"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/peer"
+ to "github.com/apache/incubator-trafficcontrol/traffic_ops/client"
)
+// CopyTrafficMonitorConfigMap returns a deep copy of the given TrafficMonitorConfigMap
func CopyTrafficMonitorConfigMap(a *to.TrafficMonitorConfigMap) to.TrafficMonitorConfigMap {
b := to.TrafficMonitorConfigMap{}
b.TrafficServer = map[string]to.TrafficServer{}
@@ -41,11 +64,13 @@ func CopyTrafficMonitorConfigMap(a *to.TrafficMonitorConfigMap) to.TrafficMonito
return b
}
+// TrafficMonitorConfigMapThreadsafe encapsulates a TrafficMonitorConfigMap safe for multiple readers and a single writer.
type TrafficMonitorConfigMapThreadsafe struct {
monitorConfig *to.TrafficMonitorConfigMap
m *sync.RWMutex
}
+// NewTrafficMonitorConfigMapThreadsafe returns an encapsulated TrafficMonitorConfigMap safe for multiple readers and a single writer.
func NewTrafficMonitorConfigMapThreadsafe() TrafficMonitorConfigMapThreadsafe {
return TrafficMonitorConfigMapThreadsafe{monitorConfig: &to.TrafficMonitorConfigMap{}, m: &sync.RWMutex{}}
}
@@ -64,12 +89,13 @@ func (t *TrafficMonitorConfigMapThreadsafe) Set(c to.TrafficMonitorConfigMap) {
t.m.Unlock()
}
+// StartMonitorConfigManager runs the monitor config manager goroutine, and returns the threadsafe data which it sets.
func StartMonitorConfigManager(
monitorConfigPollChan <-chan to.TrafficMonitorConfigMap,
localStates peer.CRStatesThreadsafe,
- statUrlSubscriber chan<- poller.HttpPollerConfig,
- healthUrlSubscriber chan<- poller.HttpPollerConfig,
- peerUrlSubscriber chan<- poller.HttpPollerConfig,
+ statURLSubscriber chan<- poller.HttpPollerConfig,
+ healthURLSubscriber chan<- poller.HttpPollerConfig,
+ peerURLSubscriber chan<- poller.HttpPollerConfig,
cachesChangeSubscriber chan<- struct{},
cfg config.Config,
staticAppData StaticAppData,
@@ -78,9 +104,9 @@ func StartMonitorConfigManager(
go monitorConfigListen(monitorConfig,
monitorConfigPollChan,
localStates,
- statUrlSubscriber,
- healthUrlSubscriber,
- peerUrlSubscriber,
+ statURLSubscriber,
+ healthURLSubscriber,
+ peerURLSubscriber,
cachesChangeSubscriber,
cfg,
staticAppData,
@@ -88,96 +114,166 @@ func StartMonitorConfigManager(
return monitorConfig
}
+// trafficOpsHealthConnectionTimeoutToDuration takes the int from Traffic Ops, which is in milliseconds, and returns a time.Duration
+// TODO change Traffic Ops Client API to a time.Duration
+func trafficOpsHealthConnectionTimeoutToDuration(t int) time.Duration {
+ return time.Duration(t) * time.Millisecond
+}
+
+// trafficOpsPeerPollIntervalToDuration takes the int from Traffic Ops, which is in milliseconds, and returns a time.Duration
+// TODO change Traffic Ops Client API to a time.Duration
+func trafficOpsPeerPollIntervalToDuration(t int) time.Duration {
+ return time.Duration(t) * time.Millisecond
+}
+
+// trafficOpsStatPollIntervalToDuration takes the int from Traffic Ops, which is in milliseconds, and returns a time.Duration
+// TODO change Traffic Ops Client API to a time.Duration
+func trafficOpsStatPollIntervalToDuration(t int) time.Duration {
+ return time.Duration(t) * time.Millisecond
+}
+
+// trafficOpsHealthPollIntervalToDuration takes the int from Traffic Ops, which is in milliseconds, and returns a time.Duration
+// TODO change Traffic Ops Client API to a time.Duration
+func trafficOpsHealthPollIntervalToDuration(t int) time.Duration {
+ return time.Duration(t) * time.Millisecond
+}
+
+// getPollIntervals reads the Traffic Ops Client monitorConfig structure, and parses and returns the health, peer, and stat poll intervals
+func getHealthPeerStatPollIntervals(monitorConfig to.TrafficMonitorConfigMap, cfg config.Config) (time.Duration, time.Duration, time.Duration, error) {
+ healthPollIntervalI, healthPollIntervalExists := monitorConfig.Config["health.polling.interval"]
+ if !healthPollIntervalExists {
+ return 0, 0, 0, fmt.Errorf("Traffic Ops Monitor config missing 'health.polling.interval', not setting config changes.\n")
+ }
+ healthPollIntervalInt, healthPollIntervalIsInt := healthPollIntervalI.(float64)
+ if !healthPollIntervalIsInt {
+ return 0, 0, 0, fmt.Errorf("Traffic Ops Monitor config 'health.polling.interval' value '%v' type %T is not an integer, not setting config changes.\n", healthPollIntervalI, healthPollIntervalI)
+ }
+ healthPollInterval := trafficOpsHealthPollIntervalToDuration(int(healthPollIntervalInt))
+
+ peerPollIntervalI, peerPollIntervalExists := monitorConfig.Config["peers.polling.interval"]
+ if !peerPollIntervalExists {
+ return 0, 0, 0, fmt.Errorf("Traffic Ops Monitor config missing 'peers.polling.interval', not setting config changes.\n")
+ }
+ peerPollIntervalInt, peerPollIntervalIsInt := peerPollIntervalI.(float64)
+ if !peerPollIntervalIsInt {
+ return 0, 0, 0, fmt.Errorf("Traffic Ops Monitor config 'peers.polling.interval' value '%v' type %T is not an integer, not setting config changes.\n", peerPollIntervalI, peerPollIntervalI)
+ }
+ peerPollInterval := trafficOpsPeerPollIntervalToDuration(int(peerPollIntervalInt))
+
+ statPollIntervalI, statPollIntervalExists := monitorConfig.Config["stat.polling.interval"]
+ if !statPollIntervalExists {
+ log.Warnf("Traffic Ops Monitor config missing 'stat.polling.interval', using health for stat.\n")
+ statPollIntervalI = healthPollIntervalI
+ }
+ statPollIntervalInt, statPollIntervalIsInt := statPollIntervalI.(float64)
+ if !statPollIntervalIsInt {
+ log.Warnf("Traffic Ops Monitor config 'stat.polling.interval' value '%v' type %T is not an integer, using health for stat\n", statPollIntervalI, statPollIntervalI)
+ statPollIntervalInt = healthPollIntervalInt
+ }
+ statPollInterval := trafficOpsStatPollIntervalToDuration(int(statPollIntervalInt))
+
+ // Formerly, only 'health' polling existed. If TO still has old configuration and doesn't have a 'stat' parameter, this allows us to assume the 'health' poll is slow, and sets it to the stat poll (which used to be the only poll, getting all astats data) to the given presumed-slow health poll, and set the now-fast-and-small health poll to a short fraction of that.
+ if healthPollIntervalExists && !statPollIntervalExists {
+ healthPollInterval = time.Duration(float64(healthPollInterval) / float64(cfg.HealthToStatRatio))
+ }
+
+ return healthPollInterval, peerPollInterval, statPollInterval, nil
+}
+
// TODO timing, and determine if the case, or its internal `for`, should be put in a goroutine
// TODO determine if subscribers take action on change, and change to mutexed objects if not.
func monitorConfigListen(
monitorConfigTS TrafficMonitorConfigMapThreadsafe,
monitorConfigPollChan <-chan to.TrafficMonitorConfigMap,
localStates peer.CRStatesThreadsafe,
- statUrlSubscriber chan<- poller.HttpPollerConfig,
- healthUrlSubscriber chan<- poller.HttpPollerConfig,
- peerUrlSubscriber chan<- poller.HttpPollerConfig,
+ statURLSubscriber chan<- poller.HttpPollerConfig,
+ healthURLSubscriber chan<- poller.HttpPollerConfig,
+ peerURLSubscriber chan<- poller.HttpPollerConfig,
cachesChangeSubscriber chan<- struct{},
cfg config.Config,
staticAppData StaticAppData,
) {
- for {
- select {
- case monitorConfig := <-monitorConfigPollChan:
- monitorConfigTS.Set(monitorConfig)
- healthUrls := map[string]string{}
- statUrls := map[string]string{}
- peerUrls := map[string]string{}
- caches := map[string]string{}
-
- for _, srv := range monitorConfig.TrafficServer {
- caches[srv.HostName] = srv.Status
-
- cacheName := enum.CacheName(srv.HostName)
-
- if srv.Status == "ONLINE" {
- localStates.SetCache(cacheName, peer.IsAvailable{IsAvailable: true})
- continue
- }
- if srv.Status == "OFFLINE" {
- localStates.SetCache(cacheName, peer.IsAvailable{IsAvailable: false})
- continue
- }
- // seed states with available = false until our polling cycle picks up a result
- if _, exists := localStates.Get().Caches[cacheName]; !exists {
- localStates.SetCache(cacheName, peer.IsAvailable{IsAvailable: false})
- }
-
- url := monitorConfig.Profile[srv.Profile].Parameters.HealthPollingURL
- r := strings.NewReplacer(
- "${hostname}", srv.FQDN,
- "${interface_name}", srv.InterfaceName,
- "application=system", "application=plugin.remap",
- "application=", "application=plugin.remap",
- )
- url = r.Replace(url)
- healthUrls[srv.HostName] = url
- r = strings.NewReplacer("application=plugin.remap", "application=")
- url = r.Replace(url)
- statUrls[srv.HostName] = url
+ for monitorConfig := range monitorConfigPollChan {
+ monitorConfigTS.Set(monitorConfig)
+ healthURLs := map[string]poller.PollConfig{}
+ statURLs := map[string]poller.PollConfig{}
+ peerURLs := map[string]poller.PollConfig{}
+ caches := map[string]string{}
+
+ healthPollInterval, peerPollInterval, statPollInterval, err := getHealthPeerStatPollIntervals(monitorConfig, cfg)
+ if err != nil {
+ continue
+ }
+
+ for _, srv := range monitorConfig.TrafficServer {
+ caches[srv.HostName] = srv.Status
+
+ cacheName := enum.CacheName(srv.HostName)
+
+ srvStatus := enum.CacheStatusFromString(srv.Status)
+ if srvStatus == enum.CacheStatusOnline {
+ localStates.SetCache(cacheName, peer.IsAvailable{IsAvailable: true})
+ continue
}
+ if srvStatus == enum.CacheStatusOffline {
+ continue
+ }
+ // seed states with available = false until our polling cycle picks up a result
+ if _, exists := localStates.Get().Caches[cacheName]; !exists {
+ localStates.SetCache(cacheName, peer.IsAvailable{IsAvailable: false})
+ }
+
+ url := monitorConfig.Profile[srv.Profile].Parameters.HealthPollingURL
+ r := strings.NewReplacer(
+ "${hostname}", srv.IP,
+ "${interface_name}", srv.InterfaceName,
+ "application=system", "application=plugin.remap",
+ "application=", "application=plugin.remap",
+ )
+ url = r.Replace(url)
+
+ connTimeout := trafficOpsHealthConnectionTimeoutToDuration(monitorConfig.Profile[srv.Profile].Parameters.HealthConnectionTimeout)
+ healthURLs[srv.HostName] = poller.PollConfig{URL: url, Timeout: connTimeout}
+ r = strings.NewReplacer("application=plugin.remap", "application=")
+ statURL := r.Replace(url)
+ statURLs[srv.HostName] = poller.PollConfig{URL: statURL, Timeout: connTimeout}
+ }
- for _, srv := range monitorConfig.TrafficMonitor {
- if srv.HostName == staticAppData.Hostname {
- continue
- }
- if srv.Status != "ONLINE" {
- continue
- }
- // TODO: the URL should be config driven. -jse
- url := fmt.Sprintf("http://%s:%d/publish/CrStates?raw", srv.IP, srv.Port)
- peerUrls[srv.HostName] = url
+ for _, srv := range monitorConfig.TrafficMonitor {
+ if srv.HostName == staticAppData.Hostname {
+ continue
}
+ if enum.CacheStatusFromString(srv.Status) != enum.CacheStatusOnline {
+ continue
+ }
+ // TODO: the URL should be config driven. -jse
+ url := fmt.Sprintf("http://%s:%d/publish/CrStates?raw", srv.IP, srv.Port)
+ peerURLs[srv.HostName] = poller.PollConfig{URL: url} // TODO determine timeout.
+ }
- statUrlSubscriber <- poller.HttpPollerConfig{Urls: statUrls, Interval: cfg.CacheStatPollingInterval}
- healthUrlSubscriber <- poller.HttpPollerConfig{Urls: healthUrls, Interval: cfg.CacheHealthPollingInterval}
- peerUrlSubscriber <- poller.HttpPollerConfig{Urls: peerUrls, Interval: cfg.PeerPollingInterval}
+ statURLSubscriber <- poller.HttpPollerConfig{Urls: statURLs, Interval: statPollInterval}
+ healthURLSubscriber <- poller.HttpPollerConfig{Urls: healthURLs, Interval: healthPollInterval}
+ peerURLSubscriber <- poller.HttpPollerConfig{Urls: peerURLs, Interval: peerPollInterval}
- for cacheName := range localStates.GetCaches() {
- if _, exists := monitorConfig.TrafficServer[string(cacheName)]; !exists {
- log.Warnf("Removing %s from localStates", cacheName)
- localStates.DeleteCache(cacheName)
- }
+ for cacheName := range localStates.GetCaches() {
+ if _, exists := monitorConfig.TrafficServer[string(cacheName)]; !exists {
+ log.Warnf("Removing %s from localStates", cacheName)
+ localStates.DeleteCache(cacheName)
}
+ }
- cachesChangeSubscriber <- struct{}{}
+ cachesChangeSubscriber <- struct{}{}
- // TODO because there are multiple writers to localStates.DeliveryService, there is a race condition, where MonitorConfig (this func) and HealthResultManager could write at the same time, and the HealthResultManager could overwrite a delivery service addition or deletion here. Probably the simplest and most performant fix would be a lock-free algorithm using atomic compare-and-swaps.
- for _, ds := range monitorConfig.DeliveryService {
- // since caches default to unavailable, also default DS false
- if _, exists := localStates.Get().Deliveryservice[enum.DeliveryServiceName(ds.XMLID)]; !exists {
- localStates.SetDeliveryService(enum.DeliveryServiceName(ds.XMLID), peer.Deliveryservice{IsAvailable: false, DisabledLocations: []enum.CacheName{}}) // important to initialize DisabledLocations, so JSON is `[]` not `null`
- }
+ // TODO because there are multiple writers to localStates.DeliveryService, there is a race condition, where MonitorConfig (this func) and HealthResultManager could write at the same time, and the HealthResultManager could overwrite a delivery service addition or deletion here. Probably the simplest and most performant fix would be a lock-free algorithm using atomic compare-and-swaps.
+ for _, ds := range monitorConfig.DeliveryService {
+ // since caches default to unavailable, also default DS false
+ if _, exists := localStates.Get().Deliveryservice[enum.DeliveryServiceName(ds.XMLID)]; !exists {
+ localStates.SetDeliveryService(enum.DeliveryServiceName(ds.XMLID), peer.Deliveryservice{IsAvailable: false, DisabledLocations: []enum.CacheName{}}) // important to initialize DisabledLocations, so JSON is `[]` not `null`
}
- for ds, _ := range localStates.Get().Deliveryservice {
- if _, exists := monitorConfig.DeliveryService[string(ds)]; !exists {
- localStates.DeleteDeliveryService(ds)
- }
+ }
+ for ds := range localStates.Get().Deliveryservice {
+ if _, exists := monitorConfig.DeliveryService[string(ds)]; !exists {
+ localStates.DeleteDeliveryService(ds)
}
}
}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/be5ab888/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 fd41a12..813399f 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/opsconfig.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/opsconfig.go
@@ -1,42 +1,68 @@
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 (
"fmt"
"sync"
"time"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/common/handler"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/common/log"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/common/poller"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/http_server"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/peer"
- todata "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/trafficopsdata"
- towrap "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/trafficopswrapper"
- to "github.com/Comcast/traffic_control/traffic_ops/client"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/common/handler"
+ "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/peer"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/srvhttp"
+ todata "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/trafficopsdata"
+ towrap "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/trafficopswrapper"
+ to "github.com/apache/incubator-trafficcontrol/traffic_ops/client"
)
+// OpsConfigThreadsafe provides safe access for multiple reader goroutines and a single writer to a stored OpsConfig object.
// This could be made lock-free, if the performance was necessary
type OpsConfigThreadsafe struct {
opsConfig *handler.OpsConfig
m *sync.RWMutex
}
+// NewOpsConfigThreadsafe returns a new single-writer-multiple-reader OpsConfig
func NewOpsConfigThreadsafe() OpsConfigThreadsafe {
return OpsConfigThreadsafe{m: &sync.RWMutex{}, opsConfig: &handler.OpsConfig{}}
}
+// Get gets the internal OpsConfig object. This MUST NOT be modified. If modification is necessary, copy the object.
func (o *OpsConfigThreadsafe) Get() handler.OpsConfig {
o.m.RLock()
defer o.m.RUnlock()
return *o.opsConfig
}
+// Set sets the internal OpsConfig object. This MUST NOT be called from multiple goroutines.
func (o *OpsConfigThreadsafe) Set(newOpsConfig handler.OpsConfig) {
o.m.Lock()
*o.opsConfig = newOpsConfig
o.m.Unlock()
}
+// StartOpsConfigManager starts the ops config manager goroutine, returning the (threadsafe) variables which it sets.
// Note the OpsConfigManager is in charge of the httpServer, because ops config changes trigger server changes. If other things needed to trigger server restarts, the server could be put in its own goroutine with signal channels
func StartOpsConfigManager(
opsConfigFile string,
@@ -49,7 +75,7 @@ func StartOpsConfigManager(
combinedStates peer.CRStatesThreadsafe,
statHistory StatHistoryThreadsafe,
lastStats LastStatsThreadsafe,
- dsStats DSStatsThreadsafe,
+ dsStats DSStatsReader,
events EventsThreadsafe,
staticAppData StaticAppData,
healthPollInterval time.Duration,
@@ -59,6 +85,8 @@ func StartOpsConfigManager(
errorCount UintThreadsafe,
localCacheStatus CacheAvailableStatusThreadsafe,
unpolledCaches UnpolledCachesThreadsafe,
+ monitorConfig TrafficMonitorConfigMapThreadsafe,
+ cfg config.Config,
) OpsConfigThreadsafe {
opsConfigFileChannel := make(chan interface{})
@@ -80,77 +108,71 @@ func StartOpsConfigManager(
// TODO remove change subscribers, give Threadsafes directly to the things that need them. If they only set vars, and don't actually do work on change.
go func() {
- httpServer := http_server.Server{}
-
- for {
- select {
- case newOpsConfig := <-opsConfigChannel:
- var err error
- opsConfig.Set(newOpsConfig)
-
- listenAddress := ":80" // default
-
- if newOpsConfig.HttpListener != "" {
- listenAddress = newOpsConfig.HttpListener
- }
-
- handleErr := func(err error) {
- errorCount.Inc()
- log.Errorf("OpsConfigManager: %v\n", err)
- }
-
- err = httpServer.Run(func(req http_server.DataRequest) ([]byte, int) {
- return DataRequest(
- req,
- opsConfig,
- toSession,
- localStates,
- peerStates,
- combinedStates,
- statHistory,
- dsStats,
- events,
- staticAppData,
- healthPollInterval,
- lastHealthDurations,
- fetchCount,
- healthIteration,
- errorCount,
- toData,
- localCacheStatus,
- lastStats,
- unpolledCaches,
- )
- }, listenAddress)
- if err != nil {
- handleErr(fmt.Errorf("MonitorConfigPoller: error creating HTTP server: %s\n", err))
- continue
- }
-
- realToSession, err := to.Login(newOpsConfig.Url, newOpsConfig.Username, newOpsConfig.Password, newOpsConfig.Insecure)
- if err != nil {
- handleErr(fmt.Errorf("MonitorConfigPoller: error instantiating Session with traffic_ops: %s\n", err))
- continue
- }
- toSession.Set(realToSession)
-
- if err := toData.Fetch(toSession, newOpsConfig.CdnName); err != nil {
- handleErr(fmt.Errorf("Error getting Traffic Ops data: %v\n", err))
- continue
- }
-
- // These must be in a goroutine, because the monitorConfigPoller tick sends to a channel this select listens for. Thus, if we block on sends to the monitorConfigPoller, we have a livelock race condition.
- // More generically, we're using goroutines as an infinite chan buffer, to avoid potential livelocks
- for _, subscriber := range opsConfigChangeSubscribers {
- go func() {
- subscriber <- newOpsConfig // this is needed for cdnName
- }()
- }
- for _, subscriber := range toChangeSubscribers {
- go func() {
- subscriber <- toSession
- }()
- }
+ httpServer := srvhttp.Server{}
+
+ for newOpsConfig := range opsConfigChannel {
+ var err error
+ opsConfig.Set(newOpsConfig)
+
+ listenAddress := ":80" // default
+
+ if newOpsConfig.HttpListener != "" {
+ listenAddress = newOpsConfig.HttpListener
+ }
+
+ handleErr := func(err error) {
+ errorCount.Inc()
+ log.Errorf("OpsConfigManager: %v\n", err)
+ }
+
+ err = httpServer.Run(func(req srvhttp.DataRequest) ([]byte, int) {
+ return DataRequest(
+ req,
+ opsConfig,
+ toSession,
+ localStates,
+ peerStates,
+ combinedStates,
+ statHistory,
+ dsStats,
+ events,
+ staticAppData,
+ healthPollInterval,
+ lastHealthDurations,
+ fetchCount,
+ healthIteration,
+ errorCount,
+ toData,
+ localCacheStatus,
+ lastStats,
+ unpolledCaches,
+ monitorConfig,
+ )
+ }, listenAddress, cfg.ServeReadTimeout, cfg.ServeWriteTimeout)
+ if err != nil {
+ handleErr(fmt.Errorf("MonitorConfigPoller: error creating HTTP server: %s\n", err))
+ continue
+ }
+
+ realToSession, err := to.Login(newOpsConfig.Url, newOpsConfig.Username, newOpsConfig.Password, newOpsConfig.Insecure)
+ if err != nil {
+ handleErr(fmt.Errorf("MonitorConfigPoller: error instantiating Session with traffic_ops: %s\n", err))
+ continue
+ }
+ toSession.Set(realToSession)
+
+ if err := toData.Fetch(toSession, newOpsConfig.CdnName); err != nil {
+ handleErr(fmt.Errorf("Error getting Traffic Ops data: %v\n", err))
+ continue
+ }
+
+ // These must be in a goroutine, because the monitorConfigPoller tick sends to a channel this select listens for. Thus, if we block on sends to the monitorConfigPoller, we have a livelock race condition.
+ // More generically, we're using goroutines as an infinite chan buffer, to avoid potential livelocks
+ for _, subscriber := range opsConfigChangeSubscribers {
+ go func(s chan<- handler.OpsConfig) { s <- newOpsConfig }(subscriber)
+ }
+ for _, subscriber := range toChangeSubscribers {
+ go func(s chan<- towrap.ITrafficOpsSession) { s <- toSession }(subscriber)
}
}
}()
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/be5ab888/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 2f1b783..7232919 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/peer.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/peer.go
@@ -1,11 +1,31 @@
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 (
"sort"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/common/log"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/enum"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/peer"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/common/log"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/enum"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/peer"
)
// StartPeerManager listens for peer results, and when it gets one, it adds it to the peerStates list, and optimistically combines the good results into combinedStates
@@ -16,13 +36,10 @@ func StartPeerManager(
) peer.CRStatesThreadsafe {
combinedStates := peer.NewCRStatesThreadsafe()
go func() {
- for {
- select {
- case crStatesResult := <-peerChan:
- peerStates.Set(crStatesResult.Id, crStatesResult.PeerStats)
- combinedStates.Set(combineCrStates(peerStates.Get(), localStates.Get()))
- crStatesResult.PollFinished <- crStatesResult.PollID
- }
+ for crStatesResult := range peerChan {
+ peerStates.Set(crStatesResult.ID, crStatesResult.PeerStats)
+ combinedStates.Set(combineCrStates(peerStates.Get(), localStates.Get()))
+ crStatesResult.PollFinished <- crStatesResult.PollID
}
}()
return combinedStates
@@ -79,6 +96,7 @@ func combineCrStates(peerStates map[enum.TrafficMonitorName]peer.Crstates, local
return combinedStates
}
+// CacheNameSlice is a slice of cache names, which fulfills the `sort.Interface` interface.
type CacheNameSlice []enum.CacheName
func (p CacheNameSlice) Len() int { return len(p) }
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/be5ab888/traffic_monitor/experimental/traffic_monitor/manager/polledcaches.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/polledcaches.go b/traffic_monitor/experimental/traffic_monitor/manager/polledcaches.go
index c1cb999..3f353c4 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/polledcaches.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/polledcaches.go
@@ -1,9 +1,29 @@
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 (
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/common/log"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/cache"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/enum"
+ "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"
"sync"
)
@@ -14,7 +34,6 @@ type UnpolledCachesThreadsafe struct {
allCaches *map[enum.CacheName]struct{}
initialized *bool
m *sync.RWMutex
- max uint64
}
// NewUnpolledCachesThreadsafe returns a new UnpolledCachesThreadsafe object.
@@ -28,14 +47,14 @@ func NewUnpolledCachesThreadsafe() UnpolledCachesThreadsafe {
}
}
-// Get returns the caches. Callers MUST NOT modify. If mutation is necessary, copy the map
+// UnpolledCaches returns a map of caches not yet polled. Callers MUST NOT modify. If mutation is necessary, copy the map
func (t *UnpolledCachesThreadsafe) UnpolledCaches() map[enum.CacheName]struct{} {
t.m.RLock()
defer t.m.RUnlock()
return *t.unpolledCaches
}
-// Set sets the internal unpolled caches map. This is only safe for one thread of execution. This MUST NOT be called from multiple threads.
+// setUnpolledCaches sets the internal unpolled caches map. This is only safe for one thread of execution. This MUST NOT be called from multiple threads.
func (t *UnpolledCachesThreadsafe) setUnpolledCaches(v map[enum.CacheName]struct{}) {
t.m.Lock()
*t.initialized = true
@@ -47,17 +66,17 @@ func (t *UnpolledCachesThreadsafe) setUnpolledCaches(v map[enum.CacheName]struct
func (t *UnpolledCachesThreadsafe) SetNewCaches(newCaches map[enum.CacheName]struct{}) {
unpolledCaches := copyCaches(t.UnpolledCaches())
allCaches := copyCaches(*t.allCaches) // not necessary to lock `allCaches`, as the single-writer is the only thing that accesses it.
- for cache, _ := range unpolledCaches {
+ for cache := range unpolledCaches {
if _, ok := newCaches[cache]; !ok {
delete(unpolledCaches, cache)
}
}
- for cache, _ := range allCaches {
+ for cache := range allCaches {
if _, ok := newCaches[cache]; !ok {
delete(allCaches, cache)
}
}
- for cache, _ := range newCaches {
+ for cache := range newCaches {
if _, ok := allCaches[cache]; !ok {
unpolledCaches[cache] = struct{}{}
allCaches[cache] = struct{}{}
@@ -67,7 +86,7 @@ func (t *UnpolledCachesThreadsafe) SetNewCaches(newCaches map[enum.CacheName]str
t.setUnpolledCaches(unpolledCaches)
}
-// AnyCachesUnpolled returns whether there are any caches marked as not polled. Also returns true if SetNewCaches() has never been called (assuming there exist caches, if this hasn't been initialized, we couldn't have polled any of them).
+// Any returns whether there are any caches marked as not polled. Also returns true if SetNewCaches() has never been called (assuming there exist caches, if this hasn't been initialized, we couldn't have polled any of them).
func (t *UnpolledCachesThreadsafe) Any() bool {
t.m.Lock()
defer t.m.Unlock()
@@ -77,7 +96,7 @@ func (t *UnpolledCachesThreadsafe) Any() bool {
// copyCaches performs a deep copy of the given map.
func copyCaches(a map[enum.CacheName]struct{}) map[enum.CacheName]struct{} {
b := map[enum.CacheName]struct{}{}
- for k, _ := range a {
+ for k := range a {
b[k] = struct{}{}
}
return b
@@ -93,10 +112,10 @@ func (t *UnpolledCachesThreadsafe) SetPolled(results []cache.Result, lastStatsTh
return
}
lastStats := lastStatsThreadsafe.Get()
- for cache, _ := range unpolledCaches {
+ for cache := range unpolledCaches {
innerLoop:
for _, result := range results {
- if result.Id != cache {
+ if result.ID != cache {
continue
}
if !result.Available || len(result.Errors) > 0 {
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/be5ab888/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 cb0c7c3..c938974 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/stathistory.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/stathistory.go
@@ -1,28 +1,49 @@
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/Comcast/traffic_control/traffic_monitor/experimental/common/log"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/cache"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/config"
- ds "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/deliveryservice"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/enum"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/peer"
- todata "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/trafficopsdata"
+ "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/peer"
+ todata "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/trafficopsdata"
+ to "github.com/apache/incubator-trafficcontrol/traffic_ops/client"
)
+// StatHistory is a map of cache names, to an array of result history from each cache.
type StatHistory map[enum.CacheName][]cache.Result
func copyStat(a []cache.Result) []cache.Result {
b := make([]cache.Result, len(a), len(a))
- for i, v := range a {
- b[i] = v
- }
+ copy(b, a)
return b
}
+// Copy copies returns a deep copy of this StatHistory
func (a StatHistory) Copy() StatHistory {
b := StatHistory{}
for k, v := range a {
@@ -31,35 +52,32 @@ func (a StatHistory) Copy() StatHistory {
return b
}
+// StatHistoryThreadsafe provides safe access for multiple goroutines readers and a single writer to a stored StatHistory object.
// This could be made lock-free, if the performance was necessary
// TODO add separate locks for Caches and Deliveryservice maps?
type StatHistoryThreadsafe struct {
statHistory *StatHistory
m *sync.RWMutex
- max uint64
-}
-
-func (h StatHistoryThreadsafe) Max() uint64 {
- return h.max
}
-func NewStatHistoryThreadsafe(maxHistory uint64) StatHistoryThreadsafe {
+// NewStatHistoryThreadsafe returns a new StatHistory safe for multiple readers and a single writer.
+func NewStatHistoryThreadsafe() StatHistoryThreadsafe {
h := StatHistory{}
- return StatHistoryThreadsafe{m: &sync.RWMutex{}, statHistory: &h, max: maxHistory}
+ return StatHistoryThreadsafe{m: &sync.RWMutex{}, statHistory: &h}
}
// Get returns the StatHistory. Callers MUST NOT modify. If mutation is necessary, call StatHistory.Copy()
-func (t *StatHistoryThreadsafe) Get() StatHistory {
- t.m.RLock()
- defer t.m.RUnlock()
- return *t.statHistory
+func (h *StatHistoryThreadsafe) Get() StatHistory {
+ h.m.RLock()
+ defer h.m.RUnlock()
+ return *h.statHistory
}
// Set sets the internal StatHistory. This is only safe for one thread of execution. This MUST NOT be called from multiple threads.
-func (t *StatHistoryThreadsafe) Set(v StatHistory) {
- t.m.Lock()
- *t.statHistory = v
- t.m.Unlock()
+func (h *StatHistoryThreadsafe) Set(v StatHistory) {
+ h.m.Lock()
+ *h.statHistory = v
+ h.m.Unlock()
}
func pruneHistory(history []cache.Result, limit uint64) []cache.Result {
@@ -95,8 +113,8 @@ func StartStatHistoryManager(
errorCount UintThreadsafe,
cfg config.Config,
monitorConfig TrafficMonitorConfigMapThreadsafe,
-) (StatHistoryThreadsafe, DurationMapThreadsafe, LastStatsThreadsafe, DSStatsThreadsafe, UnpolledCachesThreadsafe) {
- statHistory := NewStatHistoryThreadsafe(cfg.MaxStatHistory)
+) (StatHistoryThreadsafe, DurationMapThreadsafe, LastStatsThreadsafe, DSStatsReader, UnpolledCachesThreadsafe) {
+ statHistory := NewStatHistoryThreadsafe()
lastStatDurations := NewDurationMapThreadsafe()
lastStatEndTimes := map[enum.CacheName]time.Time{}
lastStats := NewLastStatsThreadsafe()
@@ -119,21 +137,21 @@ func StartStatHistoryManager(
unpolledCaches.SetNewCaches(getNewCaches(localStates, monitorConfig))
case <-tick:
log.Warnf("StatHistoryManager flushing queued results\n")
- processStatResults(results, statHistory, combinedStates.Get(), lastStats, toData.Get(), errorCount, dsStats, lastStatEndTimes, lastStatDurations, unpolledCaches)
+ processStatResults(results, statHistory, combinedStates.Get(), lastStats, toData.Get(), errorCount, dsStats, lastStatEndTimes, lastStatDurations, unpolledCaches, monitorConfig.Get())
break innerLoop
default:
select {
case r := <-cacheStatChan:
results = append(results, r)
default:
- processStatResults(results, statHistory, combinedStates.Get(), lastStats, toData.Get(), errorCount, dsStats, lastStatEndTimes, lastStatDurations, unpolledCaches)
+ processStatResults(results, statHistory, combinedStates.Get(), lastStats, toData.Get(), errorCount, dsStats, lastStatEndTimes, lastStatDurations, unpolledCaches, monitorConfig.Get())
break innerLoop
}
}
}
}
}()
- return statHistory, lastStatDurations, lastStats, dsStats, unpolledCaches
+ return statHistory, lastStatDurations, lastStats, &dsStats, unpolledCaches
}
// 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.
@@ -148,12 +166,13 @@ func processStatResults(
lastStatEndTimes map[enum.CacheName]time.Time,
lastStatDurationsThreadsafe DurationMapThreadsafe,
unpolledCaches UnpolledCachesThreadsafe,
+ mc to.TrafficMonitorConfigMap,
) {
statHistory := statHistoryThreadsafe.Get().Copy()
- maxStats := statHistoryThreadsafe.Max()
for _, result := range results {
+ maxStats := uint64(mc.Profile[mc.TrafficServer[string(result.ID)].Profile].Parameters.HistoryCount)
// TODO determine if we want to add results with errors, or just print the errors now and don't add them.
- statHistory[enum.CacheName(result.Id)] = pruneHistory(append(statHistory[enum.CacheName(result.Id)], result), maxStats)
+ statHistory[result.ID] = pruneHistory(append(statHistory[result.ID], result), maxStats)
}
statHistoryThreadsafe.Set(statHistory)
@@ -178,11 +197,11 @@ func processStatResults(
endTime := time.Now()
lastStatDurations := lastStatDurationsThreadsafe.Get().Copy()
for _, result := range results {
- if lastStatStart, ok := lastStatEndTimes[enum.CacheName(result.Id)]; ok {
+ if lastStatStart, ok := lastStatEndTimes[result.ID]; ok {
d := time.Since(lastStatStart)
- lastStatDurations[enum.CacheName(result.Id)] = d
+ lastStatDurations[result.ID] = d
}
- lastStatEndTimes[enum.CacheName(result.Id)] = endTime
+ lastStatEndTimes[result.ID] = endTime
// log.Debugf("poll %v %v statfinish\n", result.PollID, endTime)
result.PollFinished <- result.PollID
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/be5ab888/traffic_monitor/experimental/traffic_monitor/manager/uintman.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/uintman.go b/traffic_monitor/experimental/traffic_monitor/manager/uintman.go
index 81558d0..903229a 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/uintman.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/uintman.go
@@ -1,26 +1,51 @@
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/atomic"
)
+// UintThreadsafe provides safe access for multiple goroutines readers and a single writer to a stored uint.
type UintThreadsafe struct {
val *uint64
}
+// NewUintThreadsafe returns a new single-writer-multiple-reader threadsafe uint
func NewUintThreadsafe() UintThreadsafe {
v := uint64(0)
return UintThreadsafe{val: &v}
}
+// Get gets the internal uint. This is safe for multiple readers
func (u *UintThreadsafe) Get() uint64 {
return atomic.LoadUint64(u.val)
}
+// Set sets the internal uint. This MUST NOT be called by multiple goroutines.
func (u *UintThreadsafe) Set(v uint64) {
atomic.StoreUint64(u.val, v)
}
+// Inc increments the internal uint64.
// TODO make sure everything using this uses the value it returns, not a separate Get
func (u *UintThreadsafe) Inc() uint64 {
return atomic.AddUint64(u.val, 1)
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/be5ab888/traffic_monitor/experimental/traffic_monitor/peer/crstates.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/peer/crstates.go b/traffic_monitor/experimental/traffic_monitor/peer/crstates.go
index 7554043..3dc3667 100644
--- a/traffic_monitor/experimental/traffic_monitor/peer/crstates.go
+++ b/traffic_monitor/experimental/traffic_monitor/peer/crstates.go
@@ -1,17 +1,40 @@
package peer
+/*
+ * 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 (
"encoding/json"
"sync"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/enum"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/enum"
)
+// Crstates includes availability data for caches and delivery services, as gathered and aggregated by this Traffic Monitor. It is designed to be served at an API endpoint primarily for Traffic Routers (Content Router) to consume.
+// TODO rename to `CRStates`
type Crstates struct {
Caches map[enum.CacheName]IsAvailable `json:"caches"`
Deliveryservice map[enum.DeliveryServiceName]Deliveryservice `json:"deliveryServices"`
}
+// NewCrstates creates a new CR states object, initializing pointer members.
func NewCrstates() Crstates {
return Crstates{
Caches: map[enum.CacheName]IsAvailable{},
@@ -19,6 +42,7 @@ func NewCrstates() Crstates {
}
}
+// Copy creates a deep copy of this object. It does not mutate, and is thus safe for multiple goroutines.
func (a Crstates) Copy() Crstates {
b := NewCrstates()
for k, v := range a.Caches {
@@ -30,6 +54,7 @@ func (a Crstates) Copy() Crstates {
return b
}
+// CopyDeliveryservices creates a deep copy of the delivery service availability data.. It does not mutate, and is thus safe for multiple goroutines.
func (a Crstates) CopyDeliveryservices() map[enum.DeliveryServiceName]Deliveryservice {
b := map[enum.DeliveryServiceName]Deliveryservice{}
for k, v := range a.Deliveryservice {
@@ -38,6 +63,7 @@ func (a Crstates) CopyDeliveryservices() map[enum.DeliveryServiceName]Deliveryse
return b
}
+// CopyCaches creates a deep copy of the cache availability data.. It does not mutate, and is thus safe for multiple goroutines.
func (a Crstates) CopyCaches() map[enum.CacheName]IsAvailable {
b := map[enum.CacheName]IsAvailable{}
for k, v := range a.Caches {
@@ -46,26 +72,30 @@ func (a Crstates) CopyCaches() map[enum.CacheName]IsAvailable {
return b
}
+// IsAvailable contains whether the given cache or delivery service is available. It is designed for JSON serialization, namely in the Traffic Monitor 1.0 API.
type IsAvailable struct {
IsAvailable bool `json:"isAvailable"`
}
+// Deliveryservice contains data about the availability of a particular delivery service, and which caches in that delivery service have been marked as unavailable.
type Deliveryservice struct {
DisabledLocations []enum.CacheName `json:"disabledLocations"`
IsAvailable bool `json:"isAvailable"`
}
+// CrstatesUnMarshall takes bytes of a JSON string, and unmarshals them into a Crstates object.
func CrstatesUnMarshall(body []byte) (Crstates, error) {
var crStates Crstates
-
err := json.Unmarshal(body, &crStates)
return crStates, err
}
+// CrstatesMarshall serializes the given Crstates into bytes.
func CrstatesMarshall(states Crstates) ([]byte, error) {
return json.Marshal(states)
}
+// CRStatesThreadsafe provides safe access for multiple goroutines to read a single Crstates object, with a single goroutine writer.
// This could be made lock-free, if the performance was necessary
// TODO add separate locks for Caches and Deliveryservice maps?
type CRStatesThreadsafe struct {
@@ -73,11 +103,13 @@ type CRStatesThreadsafe struct {
m *sync.RWMutex
}
+// NewCRStatesThreadsafe creates a new CRStatesThreadsafe object safe for multiple goroutine readers and a single writer.
func NewCRStatesThreadsafe() CRStatesThreadsafe {
crs := NewCrstates()
return CRStatesThreadsafe{m: &sync.RWMutex{}, crStates: &crs}
}
+// Get returns the internal Crstates object for reading.
// TODO add GetCaches, GetDeliveryservices?
func (t *CRStatesThreadsafe) Get() Crstates {
t.m.RLock()
@@ -85,6 +117,7 @@ func (t *CRStatesThreadsafe) Get() Crstates {
return t.crStates.Copy()
}
+// GetDeliveryServices returns the internal Crstates delivery services map for reading.
// TODO add GetCaches, GetDeliveryservices?
func (t *CRStatesThreadsafe) GetDeliveryServices() map[enum.DeliveryServiceName]Deliveryservice {
t.m.RLock()
@@ -92,70 +125,82 @@ func (t *CRStatesThreadsafe) GetDeliveryServices() map[enum.DeliveryServiceName]
return t.crStates.CopyDeliveryservices()
}
+// GetCache returns the availability data of the given cache. This does not mutate, and is thus safe for multiple goroutines to call.
func (t *CRStatesThreadsafe) GetCache(name enum.CacheName) IsAvailable {
t.m.RLock()
defer t.m.RUnlock()
return t.crStates.Caches[name]
}
+// GetCaches returns the availability data of all caches. This does not mutate, and is thus safe for multiple goroutines to call.
func (t *CRStatesThreadsafe) GetCaches() map[enum.CacheName]IsAvailable {
t.m.RLock()
defer t.m.RUnlock()
return t.crStates.CopyCaches()
}
+// GetDeliveryService returns the availability data of the given delivery service. This does not mutate, and is thus safe for multiple goroutines to call.
func (t *CRStatesThreadsafe) GetDeliveryService(name enum.DeliveryServiceName) Deliveryservice {
t.m.RLock()
defer t.m.RUnlock()
return t.crStates.Deliveryservice[name]
}
-func (o *CRStatesThreadsafe) Set(newCRStates Crstates) {
- o.m.Lock()
- *o.crStates = newCRStates
- o.m.Unlock()
+// Set sets the internal Crstates data. This MUST NOT be called by multiple goroutines.
+func (t *CRStatesThreadsafe) Set(newCRStates Crstates) {
+ t.m.Lock()
+ *t.crStates = newCRStates
+ t.m.Unlock()
}
-func (o *CRStatesThreadsafe) SetCache(cacheName enum.CacheName, available IsAvailable) {
- o.m.Lock()
- o.crStates.Caches[cacheName] = available
- o.m.Unlock()
+// SetCache sets the internal availability data for a particular cache. This MUST NOT be called by multiple goroutines.
+func (t *CRStatesThreadsafe) SetCache(cacheName enum.CacheName, available IsAvailable) {
+ t.m.Lock()
+ t.crStates.Caches[cacheName] = available
+ t.m.Unlock()
}
-func (o *CRStatesThreadsafe) DeleteCache(name enum.CacheName) {
- o.m.Lock()
- delete(o.crStates.Caches, name)
- o.m.Unlock()
+// DeleteCache deletes the given cache from the internal data. This MUST NOT be called by multiple goroutines.
+func (t *CRStatesThreadsafe) DeleteCache(name enum.CacheName) {
+ t.m.Lock()
+ delete(t.crStates.Caches, name)
+ t.m.Unlock()
}
-func (o *CRStatesThreadsafe) SetDeliveryService(name enum.DeliveryServiceName, ds Deliveryservice) {
- o.m.Lock()
- o.crStates.Deliveryservice[name] = ds
- o.m.Unlock()
+// SetDeliveryService sets the availability data for the given delivery service. This MUST NOT be called by multiple goroutines.
+func (t *CRStatesThreadsafe) SetDeliveryService(name enum.DeliveryServiceName, ds Deliveryservice) {
+ t.m.Lock()
+ t.crStates.Deliveryservice[name] = ds
+ t.m.Unlock()
}
-func (o *CRStatesThreadsafe) SetDeliveryServices(deliveryServices map[enum.DeliveryServiceName]Deliveryservice) {
- o.m.Lock()
- o.crStates.Deliveryservice = deliveryServices
- o.m.Unlock()
+// SetDeliveryServices sets the availability data for all delivery service. This MUST NOT be called by multiple goroutines.
+func (t *CRStatesThreadsafe) SetDeliveryServices(deliveryServices map[enum.DeliveryServiceName]Deliveryservice) {
+ t.m.Lock()
+ t.crStates.Deliveryservice = deliveryServices
+ t.m.Unlock()
}
-func (o *CRStatesThreadsafe) DeleteDeliveryService(name enum.DeliveryServiceName) {
- o.m.Lock()
- delete(o.crStates.Deliveryservice, name)
- o.m.Unlock()
+// DeleteDeliveryService deletes the given delivery service from the internal data. This MUST NOT be called by multiple goroutines.
+func (t *CRStatesThreadsafe) DeleteDeliveryService(name enum.DeliveryServiceName) {
+ t.m.Lock()
+ delete(t.crStates.Deliveryservice, name)
+ t.m.Unlock()
}
+// CRStatesPeersThreadsafe provides safe access for multiple goroutines to read a map of Traffic Monitor peers to their returned Crstates, with a single goroutine writer.
// This could be made lock-free, if the performance was necessary
type CRStatesPeersThreadsafe struct {
crStates map[enum.TrafficMonitorName]Crstates
m *sync.RWMutex
}
+// NewCRStatesPeersThreadsafe creates a new CRStatesPeers object safe for multiple goroutine readers and a single writer.
func NewCRStatesPeersThreadsafe() CRStatesPeersThreadsafe {
return CRStatesPeersThreadsafe{m: &sync.RWMutex{}, crStates: map[enum.TrafficMonitorName]Crstates{}}
}
+// Get returns the internal Traffic Monitor peer Crstates data. This MUST NOT be modified.
func (t *CRStatesPeersThreadsafe) Get() map[enum.TrafficMonitorName]Crstates {
t.m.RLock()
m := map[enum.TrafficMonitorName]Crstates{}
@@ -166,8 +211,9 @@ func (t *CRStatesPeersThreadsafe) Get() map[enum.TrafficMonitorName]Crstates {
return m
}
-func (o *CRStatesPeersThreadsafe) Set(peerName enum.TrafficMonitorName, peerState Crstates) {
- o.m.Lock()
- o.crStates[peerName] = peerState
- o.m.Unlock()
+// Set sets the internal Traffic Monitor peer Crstates data. This MUST NOT be called by multiple goroutines.
+func (t *CRStatesPeersThreadsafe) Set(peerName enum.TrafficMonitorName, peerState Crstates) {
+ t.m.Lock()
+ t.crStates[peerName] = peerState
+ t.m.Unlock()
}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/be5ab888/traffic_monitor/experimental/traffic_monitor/peer/peer.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/peer/peer.go b/traffic_monitor/experimental/traffic_monitor/peer/peer.go
index 5c9ca90..c8958c1 100644
--- a/traffic_monitor/experimental/traffic_monitor/peer/peer.go
+++ b/traffic_monitor/experimental/traffic_monitor/peer/peer.go
@@ -1,23 +1,47 @@
package peer
+/*
+ * 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 (
"encoding/json"
"io"
+ "time"
- "github.com/Comcast/traffic_control/traffic_monitor/experimental/traffic_monitor/enum"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/traffic_monitor/enum"
)
+// Handler handles peer Traffic Monitor data, taking a raw reader, parsing the data, and passing a result object to the ResultChannel. This fulfills the common `Handler` interface.
type Handler struct {
ResultChannel chan Result
Notify int
}
+// NewHandler returns a new peer Handler.
func NewHandler() Handler {
return Handler{ResultChannel: make(chan Result)}
}
+// Result contains the data parsed from polling a peer Traffic Monitor.
type Result struct {
- Id enum.TrafficMonitorName
+ ID enum.TrafficMonitorName
Available bool
Errors []error
PeerStats Crstates
@@ -25,18 +49,13 @@ type Result struct {
PollFinished chan<- uint64
}
-const (
- NOTIFY_NEVER = iota
- NOTIFY_CHANGE
- NOTIFY_ALWAYS
-)
-
-func (handler Handler) Handle(id string, r io.Reader, err error, pollId uint64, pollFinished chan<- uint64) {
+// Handle handles a response from a polled Traffic Monitor peer, parsing the data and forwarding it to the ResultChannel.
+func (handler Handler) Handle(id string, r io.Reader, reqTime time.Duration, err error, pollID uint64, pollFinished chan<- uint64) {
result := Result{
- Id: enum.TrafficMonitorName(id),
+ ID: enum.TrafficMonitorName(id),
Available: false,
Errors: []error{},
- PollID: pollId,
+ PollID: pollID,
PollFinished: pollFinished,
}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/be5ab888/traffic_monitor/experimental/traffic_monitor/peer/peer_test.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/peer/peer_test.go b/traffic_monitor/experimental/traffic_monitor/peer/peer_test.go
index 416487f..627a825 100644
--- a/traffic_monitor/experimental/traffic_monitor/peer/peer_test.go
+++ b/traffic_monitor/experimental/traffic_monitor/peer/peer_test.go
@@ -1,5 +1,25 @@
package peer
+/*
+ * 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"
"io/ioutil"
@@ -13,7 +33,7 @@ func TestCrStates(t *testing.T) {
if err != nil {
t.Log(err)
}
- crStates, err := CrStatesUnMarshall(text)
+ crStates, err := CrstatesUnMarshall(text)
if err != nil {
t.Log(err)
}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/be5ab888/traffic_monitor/experimental/traffic_monitor/srvhttp/srvhttp.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/srvhttp/srvhttp.go b/traffic_monitor/experimental/traffic_monitor/srvhttp/srvhttp.go
new file mode 100644
index 0000000..0b93493
--- /dev/null
+++ b/traffic_monitor/experimental/traffic_monitor/srvhttp/srvhttp.go
@@ -0,0 +1,353 @@
+package srvhttp
+
+/*
+ * 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"
+ "github.com/apache/incubator-trafficcontrol/traffic_monitor/experimental/common/log"
+ "github.com/hydrogen18/stoppableListener"
+ "io/ioutil"
+ "net"
+ "net/http"
+ "net/url"
+ "sync"
+ "time"
+)
+
+// GetCommonAPIData calculates and returns API data common to most endpoints
+func GetCommonAPIData(params url.Values, t time.Time) CommonAPIData {
+ return CommonAPIData{
+ QueryParams: ParametersStr(params),
+ DateStr: DateStr(t),
+ }
+}
+
+// CommonAPIData contains generic data common to most endpoints.
+type CommonAPIData struct {
+ QueryParams string `json:"pp"`
+ DateStr string `json:"date"`
+}
+
+// Server is a re-runnable HTTP server. Server.Run() may be called repeatedly, and
+// each time the previous running server will be stopped, and the server will be
+// restarted with the new port address and data request channel.
+type Server struct {
+ getData GetDataFunc
+ stoppableListener *stoppableListener.StoppableListener
+ stoppableListenerWaitGroup sync.WaitGroup
+}
+
+// endpoints returns a map of HTTP paths to functions.
+// This is a function because Go doesn't have constant map literals.
+func (s Server) endpoints() (map[string]http.HandlerFunc, error) {
+ handleRoot, err := s.handleRootFunc()
+ handleSortableJs, err := s.handleSortableFunc()
+ if err != nil {
+ return nil, fmt.Errorf("Error getting root endpoint: %v", err)
+ }
+
+ // note: with the trailing slash, any non-trailing slash requests will get a 301 redirect
+ return map[string]http.HandlerFunc{
+ "/publish/CacheStats/": s.dataRequestFunc(CacheStats),
+ "/publish/CacheStats": s.dataRequestFunc(CacheStats),
+ "/publish/CrConfig/": s.dataRequestFunc(TRConfig),
+ "/publish/CrConfig": s.dataRequestFunc(TRConfig),
+ "/publish/CrStates/": s.handleCrStatesFunc(),
+ "/publish/CrStates": s.handleCrStatesFunc(),
+ "/publish/DsStats/": s.dataRequestFunc(DSStats),
+ "/publish/DsStats": s.dataRequestFunc(DSStats),
+ "/publish/EventLog/": s.dataRequestFunc(EventLog),
+ "/publish/EventLog": s.dataRequestFunc(EventLog),
+ "/publish/PeerStates/": s.dataRequestFunc(PeerStates),
+ "/publish/PeerStates": s.dataRequestFunc(PeerStates),
+ "/publish/StatSummary/": s.dataRequestFunc(StatSummary),
+ "/publish/StatSummary": s.dataRequestFunc(StatSummary),
+ "/publish/Stats/": s.dataRequestFunc(Stats),
+ "/publish/Stats": s.dataRequestFunc(Stats),
+ "/publish/ConfigDoc/": s.dataRequestFunc(ConfigDoc),
+ "/publish/ConfigDoc": s.dataRequestFunc(ConfigDoc),
+ "/api/cache-count/": s.dataRequestFunc(APICacheCount),
+ "/api/cache-count": s.dataRequestFunc(APICacheCount),
+ "/api/cache-available-count/": s.dataRequestFunc(APICacheAvailableCount),
+ "/api/cache-available-count": s.dataRequestFunc(APICacheAvailableCount),
+ "/api/cache-down-count/": s.dataRequestFunc(APICacheDownCount),
+ "/api/cache-down-count": s.dataRequestFunc(APICacheDownCount),
+ "/api/version/": s.dataRequestFunc(APIVersion),
+ "/api/version": s.dataRequestFunc(APIVersion),
+ "/api/traffic-ops-uri/": s.dataRequestFunc(APITrafficOpsURI),
+ "/api/traffic-ops-uri": s.dataRequestFunc(APITrafficOpsURI),
+ "/api/cache-statuses/": s.dataRequestFunc(APICacheStates),
+ "/api/cache-statuses": s.dataRequestFunc(APICacheStates),
+ "/api/bandwidth-kbps/": s.dataRequestFunc(APIBandwidthKbps),
+ "/api/bandwidth-kbps": s.dataRequestFunc(APIBandwidthKbps),
+ "/api/bandwidth-capacity-kbps/": s.dataRequestFunc(APIBandwidthCapacityKbps),
+ "/api/bandwidth-capacity-kbps": s.dataRequestFunc(APIBandwidthCapacityKbps),
+ "/": handleRoot,
+ "/sorttable.js": handleSortableJs,
+ }, nil
+}
+
+func (s Server) registerEndpoints(sm *http.ServeMux) error {
+ endpoints, err := s.endpoints()
+ if err != nil {
+ return err
+ }
+ for path, f := range endpoints {
+ sm.HandleFunc(path, f)
+ }
+ return nil
+}
+
+// Run runs a new HTTP service at the given addr, making data requests to the given c.
+// Run may be called repeatedly, and each time, will shut down any existing service first.
+// Run is NOT threadsafe, and MUST NOT be called concurrently by multiple goroutines.
+func (s Server) Run(f GetDataFunc, addr string, readTimeout time.Duration, writeTimeout time.Duration) error {
+ // TODO make an object, which itself is not threadsafe, but which encapsulates all data so multiple
+ // objects can be created and Run.
+
+ if s.stoppableListener != nil {
+ log.Infof("Stopping Web Server\n")
+ s.stoppableListener.Stop()
+ s.stoppableListenerWaitGroup.Wait()
+ }
+ log.Infof("Starting Web Server\n")
+
+ var err error
+ var originalListener net.Listener
+ if originalListener, err = net.Listen("tcp", addr); err != nil {
+ return err
+ }
+ if s.stoppableListener, err = stoppableListener.New(originalListener); err != nil {
+ return err
+ }
+
+ s.getData = f
+
+ sm := http.NewServeMux()
+ err = s.registerEndpoints(sm)
+ if err != nil {
+ return err
+ }
+ server := &http.Server{
+ Addr: addr,
+ Handler: sm,
+ ReadTimeout: readTimeout,
+ WriteTimeout: writeTimeout,
+ MaxHeaderBytes: 1 << 20,
+ }
+
+ s.stoppableListenerWaitGroup = sync.WaitGroup{}
+ s.stoppableListenerWaitGroup.Add(1)
+ go func() {
+ defer s.stoppableListenerWaitGroup.Done()
+ err := server.Serve(s.stoppableListener)
+ if err != nil {
+ log.Warnf("HTTP server stopped with error: %v\n", err)
+ }
+ }()
+
+ log.Infof("Web server listening on %s", addr)
+ return nil
+}
+
+// Type is the API request type which was received.
+type Type int
+
+const (
+ // TRConfig represents a data request for the Traffic Router config
+ TRConfig Type = (1 << iota)
+ // TRStateDerived represents a data request for the derived data, aggregated from all Traffic Monitor peers.
+ TRStateDerived
+ // TRStateSelf represents a data request for the cache health data only from this Traffic Monitor, not from its peers.
+ TRStateSelf
+ // CacheStats represents a data request for general cache stats
+ CacheStats
+ // DSStats represents a data request for delivery service stats
+ DSStats
+ // EventLog represents a data request for the event log
+ EventLog
+ // PeerStates represents a data request for the cache health data gathered from Traffic Monitor peers.
+ PeerStates
+ // StatSummary represents a data request for a summary of the gathered stats
+ StatSummary
+ // Stats represents a data request for stats
+ Stats
+ // ConfigDoc represents a data request for this app's configuration data.
+ ConfigDoc
+ // APICacheCount represents a data request for the total number of caches this Traffic Monitor polls, as received Traffic Ops.
+ APICacheCount
+ // APICacheAvailableCount represents a data request for the number of caches flagged as available by this Traffic Monitor
+ APICacheAvailableCount
+ // APICacheDownCount represents a data request for the number of caches flagged as unavailable by this Traffic Monitor
+ APICacheDownCount
+ // APIVersion represents a data request for this app's version
+ APIVersion
+ // APITrafficOpsURI represents a data request for the Traffic Ops URI this app is configured to query
+ APITrafficOpsURI
+ // APICacheStates represents a data request for a summary of the cache states
+ APICacheStates
+ // APIBandwidthKbps represents a data request for the total bandwidth of all caches polled
+ APIBandwidthKbps
+ // APIBandwidthCapacityKbps represents a data request for the total bandwidth capacity of all caches polled
+ APIBandwidthCapacityKbps
+)
+
+// String returns a string representation of the API request type.
+func (t Type) String() string {
+ switch t {
+ case TRConfig:
+ return "TRConfig"
+ case TRStateDerived:
+ return "TRStateDerived"
+ case TRStateSelf:
+ return "TRStateSelf"
+ case CacheStats:
+ return "CacheStats"
+ case DSStats:
+ return "DSStats"
+ case EventLog:
+ return "EventLog"
+ case PeerStates:
+ return "PeerStates"
+ case StatSummary:
+ return "StatSummary"
+ case Stats:
+ return "Stats"
+ case ConfigDoc:
+ return "ConfigDoc"
+ case APICacheCount:
+ return "APICacheCount"
+ case APICacheAvailableCount:
+ return "APICacheAvailableCount"
+ case APICacheDownCount:
+ return "APICacheDownCount"
+ case APIVersion:
+ return "APIVersion"
+ case APITrafficOpsURI:
+ return "APITrafficOpsURI"
+ case APICacheStates:
+ return "APICacheStates"
+ case APIBandwidthKbps:
+ return "APIBandwidthKbps"
+ case APIBandwidthCapacityKbps:
+ return "APIBandwidthCapacityKbps"
+ default:
+ return "Invalid"
+ }
+}
+
+// Format is the format protocol the API response will be.
+type Format int
+
+const (
+ // XML represents that data should be serialized to XML
+ XML Format = (1 << iota)
+ // JSON represents that data should be serialized to JSON
+ JSON
+)
+
+// DataRequest contains all the data about an API request necessary to form a response.
+type DataRequest struct {
+ Type
+ Format
+ Date string
+ Parameters map[string][]string
+}
+
+// GetDataFunc is a function which takes a DataRequest from a request made by a client, and returns the proper response to send to the client.
+type GetDataFunc func(DataRequest) ([]byte, int)
+
+// ParametersStr takes the URL query parameters, and returns a string as used by the Traffic Monitor 1.0 endpoints "pp" key.
+func ParametersStr(params url.Values) string {
+ pp := ""
+ for param, vals := range params {
+ for _, val := range vals {
+ pp += param + "=[" + val + "], "
+ }
+ }
+ if len(pp) > 2 {
+ pp = pp[:len(pp)-2]
+ }
+ return pp
+}
+
+// DateStr returns the given time in the format expected by Traffic Monitor 1.0 API users
+func DateStr(t time.Time) string {
+ return t.UTC().Format("Mon Jan 02 15:04:05 UTC 2006")
+}
+
+func (s Server) dataRequest(w http.ResponseWriter, req *http.Request, t Type, f Format) {
+ //pp: "0=[my-ats-edge-cache-0], hc=[1]",
+ //dateLayout := "Thu Oct 09 20:28:36 UTC 2014"
+ dateLayout := "Mon Jan 02 15:04:05 MST 2006"
+ data, responseCode := s.getData(DataRequest{
+ Type: t,
+ Format: f,
+ Date: time.Now().UTC().Format(dateLayout),
+ Parameters: req.URL.Query(),
+ })
+ if len(data) > 0 {
+ w.WriteHeader(responseCode)
+ if _, err := w.Write(data); err != nil {
+ log.Warnf("received error writing data request %v: %v\n", t, err)
+ }
+
+ } else {
+ w.WriteHeader(http.StatusInternalServerError)
+ if _, err := w.Write([]byte("Internal Server Error")); err != nil {
+ log.Warnf("received error writing data request %v: %v\n", t, err)
+ }
+ }
+}
+
+func (s Server) handleRootFunc() (http.HandlerFunc, error) {
+ return s.handleFile("index.html")
+}
+
+func (s Server) handleSortableFunc() (http.HandlerFunc, error) {
+ return s.handleFile("sorttable.js")
+}
+
+func (s Server) handleFile(name string) (http.HandlerFunc, error) {
+ index, err := ioutil.ReadFile(name)
+ if err != nil {
+ return nil, err
+ }
+ return func(w http.ResponseWriter, req *http.Request) {
+ fmt.Fprintf(w, "%s", index)
+ }, nil
+}
+
+func (s Server) handleCrStatesFunc() http.HandlerFunc {
+ return func(w http.ResponseWriter, req *http.Request) {
+ t := TRStateDerived
+ if req.URL.RawQuery == "raw" {
+ t = TRStateSelf
+ }
+ s.dataRequest(w, req, t, JSON)
+ }
+}
+
+func (s Server) dataRequestFunc(t Type) http.HandlerFunc {
+ return func(w http.ResponseWriter, r *http.Request) {
+ s.dataRequest(w, r, t, JSON)
+ }
+}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/be5ab888/traffic_monitor/experimental/traffic_monitor/traffic_monitor-example-config.json
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/traffic_monitor-example-config.json b/traffic_monitor/experimental/traffic_monitor/traffic_monitor-example-config.json
index 9493dd2..d1b4613 100644
--- a/traffic_monitor/experimental/traffic_monitor/traffic_monitor-example-config.json
+++ b/traffic_monitor/experimental/traffic_monitor/traffic_monitor-example-config.json
@@ -7,10 +7,12 @@
"max_events": 200,
"max_stat_history": 5,
"max_health_history": 5,
- "health_flush_interval_ms": 200,
- "stat_flush_interval_ms": 200,
+ "health_flush_interval_ms": 20,
+ "stat_flush_interval_ms": 20,
"log_location_error": "stderr",
"log_location_warning": "stdout",
"log_location_info": "null",
- "log_location_debug": "null"
+ "log_location_debug": "null",
+ "serve_read_timeout_ms": 10000,
+ "serve_write_timeout_ms": 10000
}