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 2016/11/14 17:09:08 UTC
[2/5] incubator-trafficcontrol git commit: Change TM2 HTTP dispatch
to use a map
Change TM2 HTTP dispatch to use a map
Changes Traffic Monitor 2.0 HTTP dispatching to give a map directly
to the srvhttp.Server, to register endpoints. This not only makes
serving faster, but makes the code less error prone, by directly
dispatching instead of using a giant switch statement.
Project: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/repo
Commit: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/commit/96ab1169
Tree: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/tree/96ab1169
Diff: http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/diff/96ab1169
Branch: refs/heads/master
Commit: 96ab11692f7ddfb6b8ae8fb38cd2a17d4c6134b6
Parents: 93d1372
Author: Robert Butts <ro...@gmail.com>
Authored: Mon Nov 7 14:54:48 2016 -0700
Committer: Dave Neuman <ne...@apache.org>
Committed: Mon Nov 14 10:08:17 2016 -0700
----------------------------------------------------------------------
.../traffic_monitor/manager/datarequest.go | 396 +++++++++++++------
.../traffic_monitor/manager/opsconfig.go | 46 ++-
.../traffic_monitor/srvhttp/srvhttp.go | 216 +---------
3 files changed, 310 insertions(+), 348 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/96ab1169/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
----------------------------------------------------------------------
diff --git a/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go b/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
index 0004b9d..750264a 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/datarequest.go
@@ -432,9 +432,223 @@ func NewPeerStateFilter(params url.Values, cacheTypes map[enum.CacheName]enum.Ca
}, nil
}
-// DataRequest takes an `srvhttp.DataRequest`, and the monitored data objects, and returns the appropriate response, and the status code.
-func DataRequest(
- req srvhttp.DataRequest,
+// HandleErr takes an error, and the request type it came from, and logs. It is ok to call with a nil error, in which case this is a no-op.
+func HandleErr(errorCount UintThreadsafe, reqPath string, err error) {
+ if err == nil {
+ return
+ }
+ errorCount.Inc()
+ log.Errorf("Request Error: %v\n", fmt.Errorf(reqPath+": %v", err))
+}
+
+// WrapErrCode takes the body, err, and log context (errorCount, reqPath). It logs and deals with any error, and returns the appropriate bytes and response code for the `srvhttp`. It notably returns InternalServerError status on any error, for security reasons.
+func WrapErrCode(errorCount UintThreadsafe, reqPath string, body []byte, err error) ([]byte, int) {
+ if err == nil {
+ return body, http.StatusOK
+ }
+ HandleErr(errorCount, reqPath, err)
+ return nil, http.StatusInternalServerError
+}
+
+// WrapBytes takes a function which cannot error and returns only bytes, and wraps it as a http.HandlerFunc. The errContext is logged if the write fails, and should be enough information to trace the problem (function name, endpoint, request parameters, etc).
+func WrapBytes(f func() []byte) http.HandlerFunc {
+ return func(w http.ResponseWriter, r *http.Request) {
+ log.Write(w, f(), r.URL.EscapedPath())
+ }
+}
+
+// WrapErr takes a function which returns bytes and an error, and wraps it as a http.HandlerFunc. If the error is nil, the bytes are written with Status OK. Else, the error is logged, and InternalServerError is returned as the response code. If you need to return a different response code (for example, StatusBadRequest), call wrapRespCode.
+func WrapErr(errorCount UintThreadsafe, f func() ([]byte, error)) http.HandlerFunc {
+ return func(w http.ResponseWriter, r *http.Request) {
+ bytes, err := f()
+ _, code := WrapErrCode(errorCount, r.URL.EscapedPath(), bytes, err)
+ w.WriteHeader(code)
+ log.Write(w, bytes, r.URL.EscapedPath())
+ }
+}
+
+// SrvFunc is a function which takes URL parameters, and returns the requested data, and a response code. Note it does not take the full http.Request, and does not have the path. SrvFunc functions should be called via dispatch, and any additional data needed should be closed via a lambda.
+// TODO split params and path into 2 separate wrappers?
+// TODO change to simply take the http.Request?
+type SrvFunc func(params url.Values, path string) ([]byte, int)
+
+// WrapParams takes a SrvFunc and wraps it as an http.HandlerFunc. Note if the SrvFunc returns 0 bytes, an InternalServerError is returned, and the response code is ignored, for security reasons. If an error response code is necessary, return bytes to that effect, for example, "Bad Request". DO NOT return informational messages regarding internal server errors; these should be logged, and only a 500 code returned to the client, for security reasons.
+func WrapParams(f SrvFunc) http.HandlerFunc {
+ return func(w http.ResponseWriter, r *http.Request) {
+ bytes, code := f(r.URL.Query(), r.URL.EscapedPath())
+ if len(bytes) > 0 {
+ w.WriteHeader(code)
+ if _, err := w.Write(bytes); err != nil {
+ log.Warnf("received error writing data request %v: %v\n", r.URL.EscapedPath(), 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", r.URL.EscapedPath(), err)
+ }
+ }
+ }
+}
+
+func srvTRConfig(opsConfig OpsConfigThreadsafe, toSession towrap.ITrafficOpsSession) ([]byte, error) {
+ cdnName := opsConfig.Get().CdnName
+ if toSession == nil {
+ return nil, fmt.Errorf("Unable to connect to Traffic Ops")
+ }
+ if cdnName == "" {
+ return nil, fmt.Errorf("No CDN Configured")
+ }
+ return toSession.CRConfigRaw(cdnName)
+}
+
+func makeWrapAll(errorCount UintThreadsafe, unpolledCaches UnpolledCachesThreadsafe) func(http.HandlerFunc) http.HandlerFunc {
+ return func(f http.HandlerFunc) http.HandlerFunc {
+ return wrapUnpolledCheck(unpolledCaches, errorCount, f)
+ }
+}
+
+func makeCrConfigHandler(wrapper func(http.HandlerFunc) http.HandlerFunc, errorCount UintThreadsafe, opsConfig OpsConfigThreadsafe, toSession towrap.ITrafficOpsSession) http.HandlerFunc {
+ return wrapper(WrapErr(errorCount, func() ([]byte, error) {
+ return srvTRConfig(opsConfig, toSession)
+ }))
+}
+
+func srvTRState(params url.Values, localStates peer.CRStatesThreadsafe, combinedStates peer.CRStatesThreadsafe) ([]byte, error) {
+ if _, raw := params["raw"]; raw {
+ return srvTRStateSelf(localStates)
+ }
+ return srvTRStateDerived(combinedStates)
+}
+
+func srvTRStateDerived(combinedStates peer.CRStatesThreadsafe) ([]byte, error) {
+ return peer.CrstatesMarshall(combinedStates.Get())
+}
+
+func srvTRStateSelf(localStates peer.CRStatesThreadsafe) ([]byte, error) {
+ return peer.CrstatesMarshall(localStates.Get())
+}
+
+// TODO remove error params, handle by returning an error? How, since we need to return a non-standard code?
+func srvCacheStats(params url.Values, errorCount UintThreadsafe, errContext string, toData todata.TODataThreadsafe, statHistory StatHistoryThreadsafe) ([]byte, int) {
+ filter, err := NewCacheStatFilter(params, toData.Get().ServerTypes)
+ if err != nil {
+ HandleErr(errorCount, errContext, err)
+ return []byte(err.Error()), http.StatusBadRequest
+ }
+ bytes, err := cache.StatsMarshall(statHistory.Get(), filter, params)
+ return WrapErrCode(errorCount, errContext, bytes, err)
+}
+
+func srvDSStats(params url.Values, errorCount UintThreadsafe, errContext string, toData todata.TODataThreadsafe, dsStats DSStatsReader) ([]byte, int) {
+ filter, err := NewDSStatFilter(params, toData.Get().DeliveryServiceTypes)
+ if err != nil {
+ HandleErr(errorCount, errContext, err)
+ return []byte(err.Error()), http.StatusBadRequest
+ }
+ bytes, err := json.Marshal(dsStats.Get().JSON(filter, params))
+ return WrapErrCode(errorCount, errContext, bytes, err)
+}
+
+func srvEventLog(events EventsThreadsafe) ([]byte, error) {
+ return json.Marshal(JSONEvents{Events: events.Get()})
+}
+
+func srvPeerStates(params url.Values, errorCount UintThreadsafe, errContext string, toData todata.TODataThreadsafe, peerStates peer.CRStatesPeersThreadsafe) ([]byte, int) {
+ filter, err := NewPeerStateFilter(params, toData.Get().ServerTypes)
+ if err != nil {
+ HandleErr(errorCount, errContext, err)
+ return []byte(err.Error()), http.StatusBadRequest
+ }
+ bytes, err := json.Marshal(createAPIPeerStates(peerStates.Get(), filter, params))
+ return WrapErrCode(errorCount, errContext, bytes, err)
+}
+
+func srvStatSummary() ([]byte, int) {
+ return nil, http.StatusNotImplemented
+}
+
+func srvStats(staticAppData StaticAppData, healthPollInterval time.Duration, lastHealthDurations DurationMapThreadsafe, fetchCount UintThreadsafe, healthIteration UintThreadsafe, errorCount UintThreadsafe) ([]byte, error) {
+ return getStats(staticAppData, healthPollInterval, lastHealthDurations.Get(), fetchCount.Get(), healthIteration.Get(), errorCount.Get())
+}
+
+func srvConfigDoc(opsConfig OpsConfigThreadsafe) ([]byte, error) {
+ opsConfigCopy := opsConfig.Get()
+ // if the password is blank, leave it blank, so callers can see it's missing.
+ if opsConfigCopy.Password != "" {
+ opsConfigCopy.Password = "*****"
+ }
+ return json.Marshal(opsConfigCopy)
+}
+
+// TODO determine if this should use peerStates
+func srvAPICacheCount(localStates peer.CRStatesThreadsafe) []byte {
+ return []byte(strconv.Itoa(len(localStates.Get().Caches)))
+}
+
+func srvAPICacheAvailableCount(localStates peer.CRStatesThreadsafe) []byte {
+ return []byte(strconv.Itoa(cacheAvailableCount(localStates.Get().Caches)))
+}
+
+func srvAPICacheDownCount(localStates peer.CRStatesThreadsafe, monitorConfig TrafficMonitorConfigMapThreadsafe) []byte {
+ return []byte(strconv.Itoa(cacheDownCount(localStates.Get().Caches, monitorConfig.Get().TrafficServer)))
+}
+
+func srvAPIVersion(staticAppData StaticAppData) []byte {
+ s := "traffic_monitor-" + staticAppData.Version + "."
+ if len(staticAppData.GitRevision) > 6 {
+ s += staticAppData.GitRevision[:6]
+ } else {
+ s += staticAppData.GitRevision
+ }
+ return []byte(s)
+}
+
+func srvAPITrafficOpsURI(opsConfig OpsConfigThreadsafe) []byte {
+ return []byte(opsConfig.Get().Url)
+}
+func srvAPICacheStates(toData todata.TODataThreadsafe, statHistory StatHistoryThreadsafe, lastHealthDurations DurationMapThreadsafe, localStates peer.CRStatesThreadsafe, lastStats LastStatsThreadsafe, localCacheStatus CacheAvailableStatusThreadsafe) ([]byte, error) {
+ return json.Marshal(createCacheStatuses(toData.Get().ServerTypes, statHistory.Get(), lastHealthDurations.Get(), localStates.Get().Caches, lastStats.Get(), localCacheStatus))
+}
+
+func srvAPIBandwidthKbps(toData todata.TODataThreadsafe, lastStats LastStatsThreadsafe) []byte {
+ serverTypes := toData.Get().ServerTypes
+ kbpsStats := lastStats.Get()
+ sum := float64(0.0)
+ for cache, data := range kbpsStats.Caches {
+ if serverTypes[cache] != enum.CacheTypeEdge {
+ continue
+ }
+ sum += data.Bytes.PerSec / ds.BytesPerKilobit
+ }
+ return []byte(fmt.Sprintf("%f", sum))
+}
+func srvAPIBandwidthCapacityKbps(statHistoryThs StatHistoryThreadsafe) []byte {
+ statHistory := statHistoryThs.Get()
+ cap := int64(0)
+ for _, results := range statHistory {
+ if len(results) == 0 {
+ continue
+ }
+ cap += results[0].MaxKbps
+ }
+ return []byte(fmt.Sprintf("%d", cap))
+}
+
+// WrapUnpolledCheck wraps an http.HandlerFunc, returning ServiceUnavailable if any caches are unpolled; else, calling the wrapped func.
+func wrapUnpolledCheck(unpolledCaches UnpolledCachesThreadsafe, errorCount UintThreadsafe, f http.HandlerFunc) http.HandlerFunc {
+ return func(w http.ResponseWriter, r *http.Request) {
+ if unpolledCaches.Any() {
+ HandleErr(errorCount, r.URL.EscapedPath(), fmt.Errorf("service still starting, some caches unpolled"))
+ w.WriteHeader(http.StatusServiceUnavailable)
+ log.Write(w, []byte("Service Unavailable"), r.URL.EscapedPath())
+ return
+ }
+ f(w, r)
+ }
+}
+
+// MakeDispatchMap returns the map of paths to http.HandlerFuncs for dispatching.
+func MakeDispatchMap(
opsConfig OpsConfigThreadsafe,
toSession towrap.ITrafficOpsSession,
localStates peer.CRStatesThreadsafe,
@@ -454,122 +668,66 @@ func DataRequest(
lastStats LastStatsThreadsafe,
unpolledCaches UnpolledCachesThreadsafe,
monitorConfig TrafficMonitorConfigMapThreadsafe,
-) ([]byte, int) {
-
- // handleErr takes an error, and the request type it came from, and logs. It is ok to call with a nil error, in which case this is a no-op.
- handleErr := func(err error) {
- if err == nil {
- return
- }
- errorCount.Inc()
- log.Errorf("Request Error: %v\n", fmt.Errorf(req.Type.String()+": %v", err))
- }
-
- // wrapErr takes the body, err, and the data request Type which has been processed. It logs and deals with any error, and returns the appropriate bytes and response code for the `srvhttp`. It notably returns InternalServerError status on any error, for security reasons.
- wrapErr := func(body []byte, err error) ([]byte, int) {
- if err == nil {
- return body, http.StatusOK
- }
- handleErr(err)
- return nil, http.StatusInternalServerError
- }
-
- if unpolledCaches.Any() {
- handleErr(fmt.Errorf("service still starting, some caches unpolled"))
- return []byte("Service Unavailable"), http.StatusServiceUnavailable
- }
-
- switch req.Type {
- case srvhttp.TRConfig:
- cdnName := opsConfig.Get().CdnName
- if toSession == nil {
- return wrapErr(nil, fmt.Errorf("Unable to connect to Traffic Ops"))
- }
- if cdnName == "" {
- return wrapErr(nil, fmt.Errorf("No CDN Configured"))
- }
- return wrapErr(toSession.CRConfigRaw(cdnName))
- case srvhttp.TRStateDerived:
- return wrapErr(peer.CrstatesMarshall(combinedStates.Get()))
- case srvhttp.TRStateSelf:
- return wrapErr(peer.CrstatesMarshall(localStates.Get()))
- case srvhttp.CacheStats:
- filter, err := NewCacheStatFilter(req.Parameters, toData.Get().ServerTypes)
- if err != nil {
- handleErr(err)
- return []byte(err.Error()), http.StatusBadRequest
- }
- return wrapErr(cache.StatsMarshall(statHistory.Get(), filter, req.Parameters))
- case srvhttp.DSStats:
- filter, err := NewDSStatFilter(req.Parameters, toData.Get().DeliveryServiceTypes)
- if err != nil {
- handleErr(err)
- return []byte(err.Error()), http.StatusBadRequest
- }
- // TODO marshall beforehand, for performance? (test to see how often requests are made)
- return wrapErr(json.Marshal(dsStats.Get().JSON(filter, req.Parameters)))
- case srvhttp.EventLog:
- return wrapErr(json.Marshal(JSONEvents{Events: events.Get()}))
- case srvhttp.PeerStates:
- filter, err := NewPeerStateFilter(req.Parameters, toData.Get().ServerTypes)
- if err != nil {
- handleErr(err)
- return []byte(err.Error()), http.StatusBadRequest
- }
- return wrapErr(json.Marshal(createAPIPeerStates(peerStates.Get(), filter, req.Parameters)))
- case srvhttp.StatSummary:
- return nil, http.StatusNotImplemented
- case srvhttp.Stats:
- return wrapErr(getStats(staticAppData, healthPollInterval, lastHealthDurations.Get(), fetchCount.Get(), healthIteration.Get(), errorCount.Get()))
- case srvhttp.ConfigDoc:
- opsConfigCopy := opsConfig.Get()
- // if the password is blank, leave it blank, so callers can see it's missing.
- if opsConfigCopy.Password != "" {
- opsConfigCopy.Password = "*****"
- }
- return wrapErr(json.Marshal(opsConfigCopy))
- case srvhttp.APICacheCount: // TODO determine if this should use peerStates
- return []byte(strconv.Itoa(len(localStates.Get().Caches))), http.StatusOK
- case srvhttp.APICacheAvailableCount:
- return []byte(strconv.Itoa(cacheAvailableCount(localStates.Get().Caches))), http.StatusOK
- case srvhttp.APICacheDownCount:
- return []byte(strconv.Itoa(cacheDownCount(localStates.Get().Caches, monitorConfig.Get().TrafficServer))), http.StatusOK
- case srvhttp.APIVersion:
- s := "traffic_monitor-" + staticAppData.Version + "."
- if len(staticAppData.GitRevision) > 6 {
- s += staticAppData.GitRevision[:6]
- } else {
- s += staticAppData.GitRevision
- }
- return []byte(s), http.StatusOK
- case srvhttp.APITrafficOpsURI:
- return []byte(opsConfig.Get().Url), http.StatusOK
- case srvhttp.APICacheStates:
- return wrapErr(json.Marshal(createCacheStatuses(toData.Get().ServerTypes, statHistory.Get(),
- lastHealthDurations.Get(), localStates.Get().Caches, lastStats.Get(), localCacheStatus)))
- case srvhttp.APIBandwidthKbps:
- serverTypes := toData.Get().ServerTypes
- kbpsStats := lastStats.Get()
- sum := float64(0.0)
- for cache, data := range kbpsStats.Caches {
- if serverTypes[cache] != enum.CacheTypeEdge {
- continue
- }
- sum += data.Bytes.PerSec / ds.BytesPerKilobit
- }
- return []byte(fmt.Sprintf("%f", sum)), http.StatusOK
- case srvhttp.APIBandwidthCapacityKbps:
- statHistory := statHistory.Get()
- cap := int64(0)
- for _, results := range statHistory {
- if len(results) == 0 {
- continue
- }
- cap += results[0].MaxKbps
- }
- return []byte(fmt.Sprintf("%d", cap)), http.StatusOK
- default:
- return wrapErr(nil, fmt.Errorf("Unknown Request Type"))
+) map[string]http.HandlerFunc {
+
+ // wrap composes all universal wrapper functions. Right now, it's only the UnpolledCheck, but there may be others later. For example, security headers.
+ wrap := func(f http.HandlerFunc) http.HandlerFunc {
+ return wrapUnpolledCheck(unpolledCaches, errorCount, f)
+ }
+
+ return map[string]http.HandlerFunc{
+ "/publish/CrConfig": wrap(WrapErr(errorCount, func() ([]byte, error) {
+ return srvTRConfig(opsConfig, toSession)
+ })),
+ "/publish/CrStates": wrap(WrapParams(func(params url.Values, path string) ([]byte, int) {
+ bytes, err := srvTRState(params, localStates, combinedStates)
+ return WrapErrCode(errorCount, path, bytes, err)
+ })),
+ "/publish/CacheStats": wrap(WrapParams(func(params url.Values, path string) ([]byte, int) {
+ return srvCacheStats(params, errorCount, path, toData, statHistory)
+ })),
+ "/publish/DsStats": wrap(WrapParams(func(params url.Values, path string) ([]byte, int) {
+ return srvDSStats(params, errorCount, path, toData, dsStats)
+ })),
+ "/publish/EventLog": wrap(WrapErr(errorCount, func() ([]byte, error) {
+ return srvEventLog(events)
+ })),
+ "/publish/PeerStates": wrap(WrapParams(func(params url.Values, path string) ([]byte, int) {
+ return srvPeerStates(params, errorCount, path, toData, peerStates)
+ })),
+ "/publish/StatSummary": wrap(WrapParams(func(params url.Values, path string) ([]byte, int) {
+ return srvStatSummary()
+ })),
+ "/publish/Stats": wrap(WrapErr(errorCount, func() ([]byte, error) {
+ return srvStats(staticAppData, healthPollInterval, lastHealthDurations, fetchCount, healthIteration, errorCount)
+ })),
+ "/publish/ConfigDoc": wrap(WrapErr(errorCount, func() ([]byte, error) {
+ return srvConfigDoc(opsConfig)
+ })),
+ "/api/cache-count": wrap(WrapBytes(func() []byte {
+ return srvAPICacheCount(localStates)
+ })),
+ "/api/cache-available-count": wrap(WrapBytes(func() []byte {
+ return srvAPICacheAvailableCount(localStates)
+ })),
+ "/api/cache-down-count": wrap(WrapBytes(func() []byte {
+ return srvAPICacheDownCount(localStates, monitorConfig)
+ })),
+ "/api/version": wrap(WrapBytes(func() []byte {
+ return srvAPIVersion(staticAppData)
+ })),
+ "/api/traffic-ops-uri": wrap(WrapBytes(func() []byte {
+ return srvAPITrafficOpsURI(opsConfig)
+ })),
+ "/api/cache-statuses": wrap(WrapErr(errorCount, func() ([]byte, error) {
+ return srvAPICacheStates(toData, statHistory, lastHealthDurations, localStates, lastStats, localCacheStatus)
+ })),
+ "/api/bandwidth-kbps": wrap(WrapBytes(func() []byte {
+ return srvAPIBandwidthKbps(toData, lastStats)
+ })),
+ "/api/bandwidth-capacity-kbps": wrap(WrapBytes(func() []byte {
+ return srvAPIBandwidthCapacityKbps(statHistory)
+ })),
}
}
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/96ab1169/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 813399f..f525181 100644
--- a/traffic_monitor/experimental/traffic_monitor/manager/opsconfig.go
+++ b/traffic_monitor/experimental/traffic_monitor/manager/opsconfig.go
@@ -125,30 +125,28 @@ func StartOpsConfigManager(
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)
+ endpoints := MakeDispatchMap(
+ opsConfig,
+ toSession,
+ localStates,
+ peerStates,
+ combinedStates,
+ statHistory,
+ dsStats,
+ events,
+ staticAppData,
+ healthPollInterval,
+ lastHealthDurations,
+ fetchCount,
+ healthIteration,
+ errorCount,
+ toData,
+ localCacheStatus,
+ lastStats,
+ unpolledCaches,
+ monitorConfig,
+ )
+ err = httpServer.Run(endpoints, listenAddress, cfg.ServeReadTimeout, cfg.ServeWriteTimeout)
if err != nil {
handleErr(fmt.Errorf("MonitorConfigPoller: error creating HTTP server: %s\n", err))
continue
http://git-wip-us.apache.org/repos/asf/incubator-trafficcontrol/blob/96ab1169/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
index 0b93493..84fdf78 100644
--- a/traffic_monitor/experimental/traffic_monitor/srvhttp/srvhttp.go
+++ b/traffic_monitor/experimental/traffic_monitor/srvhttp/srvhttp.go
@@ -50,79 +50,34 @@ type CommonAPIData struct {
// 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) {
+func (s Server) registerEndpoints(sm *http.ServeMux, 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)
+ return 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()
+ handleSortableJs, err := s.handleSortableFunc()
if err != nil {
- return err
+ return fmt.Errorf("Error getting sortable endpoint: %v", err)
}
+
for path, f := range endpoints {
sm.HandleFunc(path, f)
}
+
+ sm.HandleFunc("/", handleRoot)
+ sm.HandleFunc("/sorttable.js", handleSortableJs)
+
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.
-
+func (s Server) Run(endpoints map[string]http.HandlerFunc, addr string, readTimeout time.Duration, writeTimeout time.Duration) error {
if s.stoppableListener != nil {
log.Infof("Stopping Web Server\n")
s.stoppableListener.Stop()
@@ -139,10 +94,8 @@ func (s Server) Run(f GetDataFunc, addr string, readTimeout time.Duration, write
return err
}
- s.getData = f
-
sm := http.NewServeMux()
- err = s.registerEndpoints(sm)
+ err = s.registerEndpoints(sm, endpoints)
if err != nil {
return err
}
@@ -168,113 +121,6 @@ func (s Server) Run(f GetDataFunc, addr string, readTimeout time.Duration, write
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 := ""
@@ -294,30 +140,6 @@ 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")
}
@@ -335,19 +157,3 @@ func (s Server) handleFile(name string) (http.HandlerFunc, error) {
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)
- }
-}