From f91b1063ab8928a8809fe0cf7d2bbab3e46fb044 Mon Sep 17 00:00:00 2001 From: Robert Butts Date: Tue, 15 May 2018 16:27:43 -0600 Subject: [PATCH 1/2] Add TO Go caches/stats endpoint --- lib/go-tc/enum.go | 6 + .../cachesstats/cachesstats.go | 319 ++++++++++++++++++ .../crconfig/edgelocations.go | 2 +- .../traffic_ops_golang/crconfig/servers.go | 6 +- .../monitoring/monitoring.go | 8 +- traffic_ops/traffic_ops_golang/routes.go | 3 + 6 files changed, 337 insertions(+), 7 deletions(-) create mode 100644 traffic_ops/traffic_ops_golang/cachesstats/cachesstats.go diff --git a/lib/go-tc/enum.go b/lib/go-tc/enum.go index cef0c72ccf..de7a4add7c 100644 --- a/lib/go-tc/enum.go +++ b/lib/go-tc/enum.go @@ -64,6 +64,12 @@ const ( CacheTypeInvalid = CacheType("") ) +const MonitorTypeName = "RASCAL" +const MonitorProfilePrefix = "RASCAL" +const RouterTypeName = "CCR" +const EdgeTypePrefix = "EDGE" +const MidTypePrefix = "MID" + func (c CacheName) String() string { return string(c) } diff --git a/traffic_ops/traffic_ops_golang/cachesstats/cachesstats.go b/traffic_ops/traffic_ops_golang/cachesstats/cachesstats.go new file mode 100644 index 0000000000..ed0d08c4ef --- /dev/null +++ b/traffic_ops/traffic_ops_golang/cachesstats/cachesstats.go @@ -0,0 +1,319 @@ +package cachesstats + +/* + * 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 ( + "database/sql" + "encoding/json" + "errors" + "net/http" + "net/url" + "strconv" + "time" + + "github.com/apache/trafficcontrol/lib/go-tc" + "github.com/apache/trafficcontrol/traffic_ops/traffic_ops_golang/api" +) + +func Get(w http.ResponseWriter, r *http.Request) { + inf, userErr, sysErr, errCode := api.NewInfo(r, nil, nil) + if userErr != nil || sysErr != nil { + api.HandleErr(w, r, inf.Tx.Tx, errCode, userErr, sysErr) + return + } + defer inf.Close() + + api.RespWriter(w, r, inf.Tx.Tx)(getCachesStats(inf.Tx.Tx)) +} + +const MonitorProxyParameter = "tm.traffic_mon_fwd_proxy" +const MonitorRequestTimeout = time.Second * 10 +const MonitorOnlineStatus = "ONLINE" + +func getCachesStats(tx *sql.Tx) ([]CacheData, error) { + monitors, err := getCDNMonitorFQDNs(tx) + if err != nil { + return nil, errors.New("getting monitors: " + err.Error()) + } + + monitorForwardProxy, monitorForwardProxyExists, err := getGlobalParam(tx, MonitorProxyParameter) + if err != nil { + return nil, errors.New("getting global monitor proxy parameter: " + err.Error()) + } + + client := &http.Client{Timeout: MonitorRequestTimeout} + if monitorForwardProxyExists { + proxyURI, err := url.Parse(monitorForwardProxy) + if err != nil { + return nil, errors.New("monitor forward proxy '" + monitorForwardProxy + "' in parameter '" + MonitorProxyParameter + "' not a URI: " + err.Error()) + } + client = &http.Client{Timeout: MonitorRequestTimeout, Transport: &http.Transport{Proxy: http.ProxyURL(proxyURI)}} + } + + cacheData, err := getCacheData(tx) + if err != nil { + return nil, errors.New("getting cache data: " + err.Error()) + } + + for cdn, monitorFQDN := range monitors { + crStates, err := getCRStates(monitorFQDN, client) + // TODO on err, try another online monitor + if err != nil { + return nil, errors.New("getting CRStates for CDN '" + string(cdn) + "' monitor '" + monitorFQDN + "': " + err.Error()) + } + cacheData = addHealth(cacheData, crStates) + + cacheStats, err := getCacheStats(monitorFQDN, client) + if err != nil { + return nil, errors.New("getting CacheStats for CDN '" + string(cdn) + "' monitor '" + monitorFQDN + "': " + err.Error()) + } + cacheData = addStats(cacheData, cacheStats) + } + cacheData = addTotals(cacheData) + return cacheData, nil +} + +// addTotals sums each cachegroup, and adds the sum to an object with the cache-specific fields set to "ALL", and the cachegroup. It then sums all cachegroups, and adds the total to an object with all fields set to "ALL". +// TODO in the next API version, add totals in their own JSON objects, not amidst the cachegroup keys. +func addTotals(data []CacheData) []CacheData { + all := "ALL" + cachegroups := map[tc.CacheGroupName]CacheData{} + total := CacheData{Profile: all, Status: all, Healthy: true, HostName: tc.CacheName(all), CacheGroup: tc.CacheGroupName(all)} + for _, d := range data { + cg := cachegroups[tc.CacheGroupName(d.CacheGroup)] + cg.CacheGroup = d.CacheGroup + cg.Connections += d.Connections + cg.KBPS += d.KBPS + cachegroups[tc.CacheGroupName(d.CacheGroup)] = cg + total.Connections += d.Connections + total.KBPS += d.KBPS + } + for _, cg := range cachegroups { + cg.Profile = all + cg.Status = all + cg.Healthy = true + cg.HostName = tc.CacheName(all) + data = append(data, cg) + } + data = append(data, total) + return data +} + +// CRStates contains the Monitor CacheStats needed by Cachedata. It is NOT the full object served by the Monitor, but only the data required by the caches stats endpoint. +type CacheStats struct { + Caches map[tc.CacheName]CacheStat `json:"caches"` +} + +type CacheStat struct { + BandwidthKBPS []CacheStatData `json:"bandwidth"` + Connections []CacheStatData `json:"ats.proxy.process.http.current_client_connections"` +} + +type CacheStatData struct { + Value int64 `json:"value,string"` +} + +func getCacheStats(monitorFQDN string, client *http.Client) (CacheStats, error) { + path := `/publish/CacheStats?stats=ats.proxy.process.http.current_client_connections,bandwidth` + resp, err := client.Get("http://" + monitorFQDN + path) + if err != nil { + return CacheStats{}, errors.New("getting CacheStats from Monitor '" + monitorFQDN + "': " + err.Error()) + } + defer resp.Body.Close() + + cs := CacheStats{} + if err := json.NewDecoder(resp.Body).Decode(&cs); err != nil { + return CacheStats{}, errors.New("decoding CacheStats from monitor '" + monitorFQDN + "': " + err.Error()) + } + return cs, nil +} + +func addStats(cacheData []CacheData, stats CacheStats) []CacheData { + if stats.Caches == nil { + return cacheData // TODO warn? + } + for i, cache := range cacheData { + stat, ok := stats.Caches[cache.HostName] + if !ok { + continue + } + if len(stat.BandwidthKBPS) > 0 { + cache.KBPS = uint64(stat.BandwidthKBPS[0].Value) + } + if len(stat.Connections) > 0 { + cache.Connections = uint64(stat.Connections[0].Value) + } + cacheData[i] = cache + } + return cacheData +} + +// CRStates contains the Monitor CRStates needed by Cachedata. It is NOT the full object served by the Monitor, but only the data required by the caches stats endpoint. +type CRStates struct { + Caches map[tc.CacheName]Available `json:"caches"` +} + +type Available struct { + IsAvailable bool `json:"isAvailable"` +} + +func getCRStates(monitorFQDN string, client *http.Client) (CRStates, error) { + path := `/publish/CrStates` + resp, err := client.Get("http://" + monitorFQDN + path) + if err != nil { + return CRStates{}, errors.New("getting CRStates from Monitor '" + monitorFQDN + "': " + err.Error()) + } + defer resp.Body.Close() + + crs := CRStates{} + if err := json.NewDecoder(resp.Body).Decode(&crs); err != nil { + return CRStates{}, errors.New("decoding CRStates from monitor '" + monitorFQDN + "': " + err.Error()) + } + return crs, nil +} + +func addHealth(cacheData []CacheData, crStates CRStates) []CacheData { + if crStates.Caches == nil { + return cacheData // TODO warn? + } + for i, cache := range cacheData { + crsCache, ok := crStates.Caches[cache.HostName] + if !ok { + continue + } + cache.Healthy = crsCache.IsAvailable + cacheData[i] = cache + } + return cacheData +} + +type CacheData struct { + HostName tc.CacheName `json:"hostname"` + CacheGroup tc.CacheGroupName `json:"cachegroup"` + Status string `json:"status"` + Profile string `json:"profile"` + IP *string `json:"ip"` + Healthy bool `json:"healthy"` + KBPS uint64 `json:"kbps"` + Connections uint64 `json:"connections"` +} + +const CacheProfilePrefixEdge = "EDGE" +const CacheProfilePrefixMID = "MID" + +// getCacheData gets the cache data from the servers table. Note this only gets from the database, and thus does not set the Healthy member. +func getCacheData(tx *sql.Tx) ([]CacheData, error) { + qry := ` +SELECT + s.host_name, + cg.name as cachegroup, + st.name as status, + p.name as profile, + s.ip_address +FROM + server s + JOIN cachegroup cg ON s.cachegroup = cg.id + JOIN status st ON s.status = st.id + JOIN profile p ON s.profile = p.id +WHERE + p.name LIKE '` + CacheProfilePrefixEdge + `%' OR p.name LIKE '` + CacheProfilePrefixMID + `%' +` + rows, err := tx.Query(qry) + if err != nil { + return nil, errors.New("querying cache data: " + err.Error()) + } + defer rows.Close() + data := []CacheData{} + for rows.Next() { + d := CacheData{} + if err := rows.Scan(&d.HostName, &d.CacheGroup, &d.Status, &d.Profile, &d.IP); err != nil { + return nil, errors.New("scanning cache data: " + err.Error()) + } + data = append(data, d) + } + return data, nil +} + +func getMonitorForwardProxy(tx *sql.Tx) (string, error) { + forwardProxy, forwardProxyExists, err := getGlobalParam(tx, MonitorProxyParameter) + if err != nil { + return "", errors.New("getting global monitor proxy parameter: " + err.Error()) + } else if !forwardProxyExists { + forwardProxy = "" + } + return forwardProxy, nil +} + +// getCDNMonitors returns an FQDN, including port, of an online monitor for each CDN. If a CDN has no online monitors, that CDN will not have an entry in the map. If a CDN has multiple online monitors, an arbitrary one will be returned. +func getCDNMonitorFQDNs(tx *sql.Tx) (map[tc.CDNName]string, error) { + qry := ` +SELECT + s.host_name, + s.domain_name, + s.tcp_port, + c.name as cdn +FROM + server s + JOIN type t ON s.type = t.id + JOIN status st ON st.id = s.status + JOIN cdn c ON c.id = s.cdn_id +WHERE + t.name = '` + tc.MonitorTypeName + `' + AND st.name = '` + MonitorOnlineStatus + `' +` + rows, err := tx.Query(qry) + if err != nil { + return nil, errors.New("querying monitors: " + err.Error()) + } + defer rows.Close() + monitors := map[tc.CDNName]string{} + for rows.Next() { + host := "" + domain := "" + port := sql.NullInt64{} + cdn := "" + if err := rows.Scan(&host, &domain, &port, &cdn); err != nil { + return nil, errors.New("scanning monitors: " + err.Error()) + } + fqdn := host + "." + domain + if port.Valid { + fqdn += ":" + strconv.FormatInt(port.Int64, 10) + } + monitors[tc.CDNName(cdn)] = fqdn + } + return monitors, nil +} + +// getGlobalParams returns the value of the global param, whether it existed, or any error +func getGlobalParam(tx *sql.Tx, name string) (string, bool, error) { + return getParam(tx, name, "global") +} + +// getGlobalParams returns the value of the param, whether it existed, or any error. +func getParam(tx *sql.Tx, name string, configFile string) (string, bool, error) { + val := "" + if err := tx.QueryRow(`SELECT value FROM parameter WHERE name = $1 AND config_file = $2`, name, configFile).Scan(&val); err != nil { + if err == sql.ErrNoRows { + return "", false, nil + } + return "", false, errors.New("Error querying global paramter '" + name + "': " + err.Error()) + } + return val, true, nil +} diff --git a/traffic_ops/traffic_ops_golang/crconfig/edgelocations.go b/traffic_ops/traffic_ops_golang/crconfig/edgelocations.go index c7f262c9d0..7ffb96d441 100644 --- a/traffic_ops/traffic_ops_golang/crconfig/edgelocations.go +++ b/traffic_ops/traffic_ops_golang/crconfig/edgelocations.go @@ -101,7 +101,7 @@ and (st.name = 'REPORTED' or st.name = 'ONLINE' or st.name = 'ADMIN_DOWN') // to keep current default behavior when localizationMethods is unset/empty, enable all current localization methods latlon.LocalizationMethods = []tc.LocalizationMethod{tc.LocalizationMethodGeo, tc.LocalizationMethodCZ, tc.LocalizationMethodDeepCZ} } - if ttype == RouterTypeName { + if ttype == tc.RouterTypeName { routerLocs[cachegroup] = latlon } else { latlon.BackupLocations.FallbackToClosest = fallbackToClosest diff --git a/traffic_ops/traffic_ops_golang/crconfig/servers.go b/traffic_ops/traffic_ops_golang/crconfig/servers.go index 3d5abe2ebd..1fbb925451 100644 --- a/traffic_ops/traffic_ops_golang/crconfig/servers.go +++ b/traffic_ops/traffic_ops_golang/crconfig/servers.go @@ -55,7 +55,7 @@ func makeCRConfigServers(cdn string, tx *sql.Tx, cdnDomain string) ( monitors := map[string]tc.CRConfigMonitor{} for host, s := range allServers { switch { - case *s.ServerType == RouterTypeName: + case *s.ServerType == tc.RouterTypeName: status := tc.CRConfigRouterStatus(*s.ServerStatus) routers[host] = tc.CRConfigRouter{ APIPort: s.APIPort, @@ -68,7 +68,7 @@ func makeCRConfigServers(cdn string, tx *sql.Tx, cdnDomain string) ( Profile: s.Profile, ServerStatus: &status, } - case *s.ServerType == MonitorTypeName: + case *s.ServerType == tc.MonitorTypeName: monitors[host] = tc.CRConfigMonitor{ FQDN: s.Fqdn, HTTPSPort: s.HttpsPort, @@ -79,7 +79,7 @@ func makeCRConfigServers(cdn string, tx *sql.Tx, cdnDomain string) ( Profile: s.Profile, ServerStatus: s.ServerStatus, } - case strings.HasPrefix(*s.ServerType, EdgeTypePrefix) || strings.HasPrefix(*s.ServerType, MidTypePrefix): + case strings.HasPrefix(*s.ServerType, tc.EdgeTypePrefix) || strings.HasPrefix(*s.ServerType, tc.MidTypePrefix): if s.RoutingDisabled == 0 { s.CRConfigTrafficOpsServer.DeliveryServices = serverDSes[tc.CacheName(host)] } diff --git a/traffic_ops/traffic_ops_golang/monitoring/monitoring.go b/traffic_ops/traffic_ops_golang/monitoring/monitoring.go index 16b7681118..f269025863 100644 --- a/traffic_ops/traffic_ops_golang/monitoring/monitoring.go +++ b/traffic_ops/traffic_ops_golang/monitoring/monitoring.go @@ -26,12 +26,14 @@ import ( "strconv" "strings" + "github.com/apache/trafficcontrol/lib/go-tc" "github.com/apache/trafficcontrol/traffic_ops/traffic_ops_golang/api" "github.com/lib/pq" ) const CacheMonitorConfigFile = "rascal.properties" + const MonitorType = "RASCAL" const RouterType = "CCR" const MonitorProfilePrefix = "RASCAL" @@ -196,7 +198,7 @@ WHERE cdn.name = $1` return nil, nil, nil, err } - if ttype.String == MonitorType { + if ttype.String == tc.MonitorTypeName { monitors = append(monitors, Monitor{ BasicServer: BasicServer{ Profile: profile.String, @@ -225,7 +227,7 @@ WHERE cdn.name = $1` Type: ttype.String, HashID: hashID.String, }) - } else if ttype.String == RouterType { + } else if ttype.String == tc.RouterTypeName { routers = append(routers, Router{ Type: ttype.String, Profile: profile.String, @@ -381,7 +383,7 @@ FROM parameter pr JOIN profile p ON p.name LIKE '%s%%' JOIN profile_parameter pp ON pp.profile = p.id and pp.parameter = pr.id WHERE pr.config_file = '%s' -`, MonitorProfilePrefix, MonitorConfigFile) +`, tc.MonitorProfilePrefix, MonitorConfigFile) rows, err := tx.Query(query) if err != nil { diff --git a/traffic_ops/traffic_ops_golang/routes.go b/traffic_ops/traffic_ops_golang/routes.go index 1930889f3d..0a98cf33a9 100644 --- a/traffic_ops/traffic_ops_golang/routes.go +++ b/traffic_ops/traffic_ops_golang/routes.go @@ -41,6 +41,7 @@ import ( "github.com/apache/trafficcontrol/traffic_ops/traffic_ops_golang/ats" "github.com/apache/trafficcontrol/traffic_ops/traffic_ops_golang/auth" "github.com/apache/trafficcontrol/traffic_ops/traffic_ops_golang/cachegroup" + "github.com/apache/trafficcontrol/traffic_ops/traffic_ops_golang/cachesstats" "github.com/apache/trafficcontrol/traffic_ops/traffic_ops_golang/cdn" "github.com/apache/trafficcontrol/traffic_ops/traffic_ops_golang/cdnfederation" "github.com/apache/trafficcontrol/traffic_ops/traffic_ops_golang/coordinate" @@ -103,6 +104,8 @@ func Routes(d ServerData) ([]Route, []RawRoute, http.Handler, error) { {1.1, http.MethodPost, `asns/?$`, api.CreateHandler(asn.GetTypeSingleton()), auth.PrivLevelOperations, Authenticated, nil}, {1.1, http.MethodDelete, `asns/{id}$`, api.DeleteHandler(asn.GetTypeSingleton()), auth.PrivLevelOperations, Authenticated, nil}, + {1.1, http.MethodGet, `caches/stats/?(\.json)?$`, cachesstats.Get, auth.PrivLevelReadOnly, Authenticated, nil}, + //CacheGroup: CRUD {1.1, http.MethodGet, `cachegroups/trimmed/?(\.json)?$`, cachegroup.GetTrimmed, auth.PrivLevelReadOnly, Authenticated, nil}, {1.1, http.MethodGet, `cachegroups/?(\.json)?$`, api.ReadHandler(cachegroup.GetTypeSingleton()), auth.PrivLevelReadOnly, Authenticated, nil}, From 544906cc4bd54431277a5d31fc91f30fc49a62ca Mon Sep 17 00:00:00 2001 From: Robert Butts Date: Wed, 30 Jan 2019 16:27:22 -0700 Subject: [PATCH 2/2] Add TO caches/stats trying all monitors Previously, the endpoint returned an error if the first monitor it tried failed. This changes it to retry all online monitors. --- .../cachesstats/cachesstats.go | 53 ++++++++++++++----- 1 file changed, 39 insertions(+), 14 deletions(-) diff --git a/traffic_ops/traffic_ops_golang/cachesstats/cachesstats.go b/traffic_ops/traffic_ops_golang/cachesstats/cachesstats.go index ed0d08c4ef..969dbf41e0 100644 --- a/traffic_ops/traffic_ops_golang/cachesstats/cachesstats.go +++ b/traffic_ops/traffic_ops_golang/cachesstats/cachesstats.go @@ -28,7 +28,9 @@ import ( "strconv" "time" + "github.com/apache/trafficcontrol/lib/go-log" "github.com/apache/trafficcontrol/lib/go-tc" + "github.com/apache/trafficcontrol/lib/go-util" "github.com/apache/trafficcontrol/traffic_ops/traffic_ops_golang/api" ) @@ -72,19 +74,42 @@ func getCachesStats(tx *sql.Tx) ([]CacheData, error) { return nil, errors.New("getting cache data: " + err.Error()) } - for cdn, monitorFQDN := range monitors { - crStates, err := getCRStates(monitorFQDN, client) - // TODO on err, try another online monitor - if err != nil { - return nil, errors.New("getting CRStates for CDN '" + string(cdn) + "' monitor '" + monitorFQDN + "': " + err.Error()) + for cdn, monitorFQDNs := range monitors { + if len(monitorFQDNs) == 0 { + log.Warnln("getCachesStats: cdn '" + string(cdn) + "' has no online monitors, skipping!") + continue } - cacheData = addHealth(cacheData, crStates) - cacheStats, err := getCacheStats(monitorFQDN, client) - if err != nil { - return nil, errors.New("getting CacheStats for CDN '" + string(cdn) + "' monitor '" + monitorFQDN + "': " + err.Error()) + success := true + errs := []error{} + for _, monitorFQDN := range monitorFQDNs { + crStates, err := getCRStates(monitorFQDN, client) + // TODO on err, try another online monitor + if err != nil { + errs = append(errs, errors.New("getting CRStates for CDN '"+string(cdn)+"' monitor '"+monitorFQDN+"': "+err.Error())) + continue + } + + cacheStats, err := getCacheStats(monitorFQDN, client) + if err != nil { + errs = append(errs, errors.New("getting CacheStats for CDN '"+string(cdn)+"' monitor '"+monitorFQDN+"': "+err.Error())) + continue + } + + cacheData = addHealth(cacheData, crStates) + cacheData = addStats(cacheData, cacheStats) + success = true + break + } + + if !success { + return nil, errors.New("getting cache stats from all monitors failed for cdn '" + string(cdn) + "': " + util.JoinErrs(errs).Error()) + } + + // if we succeeded, log the monitor failures but don't return them + for _, err := range errs { + log.Errorln(err.Error()) } - cacheData = addStats(cacheData, cacheStats) } cacheData = addTotals(cacheData) return cacheData, nil @@ -262,7 +287,7 @@ func getMonitorForwardProxy(tx *sql.Tx) (string, error) { } // getCDNMonitors returns an FQDN, including port, of an online monitor for each CDN. If a CDN has no online monitors, that CDN will not have an entry in the map. If a CDN has multiple online monitors, an arbitrary one will be returned. -func getCDNMonitorFQDNs(tx *sql.Tx) (map[tc.CDNName]string, error) { +func getCDNMonitorFQDNs(tx *sql.Tx) (map[tc.CDNName][]string, error) { qry := ` SELECT s.host_name, @@ -283,12 +308,12 @@ WHERE return nil, errors.New("querying monitors: " + err.Error()) } defer rows.Close() - monitors := map[tc.CDNName]string{} + monitors := map[tc.CDNName][]string{} for rows.Next() { host := "" domain := "" port := sql.NullInt64{} - cdn := "" + cdn := tc.CDNName("") if err := rows.Scan(&host, &domain, &port, &cdn); err != nil { return nil, errors.New("scanning monitors: " + err.Error()) } @@ -296,7 +321,7 @@ WHERE if port.Valid { fqdn += ":" + strconv.FormatInt(port.Int64, 10) } - monitors[tc.CDNName(cdn)] = fqdn + monitors[cdn] = append(monitors[cdn], fqdn) } return monitors, nil }