Merge pull request #56 from github/cross-cluster-errors

looking into cross-cluster correlated errors

Author: Shlomi Noach

go/haproxy/parser.go

@@ -5,8 +5,13 @@ import (
 	"io/ioutil"
 	"net/http"
 	"strings"
+	"time"
+
+	"github.com/patrickmn/go-cache"
 )
 
+var csvCache = cache.New(time.Second, time.Second)
+
 // parseHeader parses the HAPRoxy CSV header, which lists column names.
 // Returned is a header-to-index map
 func parseHeader(header string) (tokensMap map[string]int) {
@@ -31,23 +36,33 @@ func parseLines(csv string) []string {
 // Such list indicates the hosts which can be expected to be active, which is then the list freno will probe.
 func ParseHosts(csvLines []string, poolName string) (hosts []string, err error) {
 	if len(csvLines) < 1 {
-		return hosts, fmt.Errorf("No lines found haproxy CSV; expecting at least a header")
+		return hosts, fmt.Errorf("Haproxy CSV parsing error: no lines found.")
+	}
+	if len(csvLines) == 1 {
+		return hosts, fmt.Errorf("Haproxy CSV parsing error: only found a header.")
 	}
 	var tokensMap map[string]int
+	poolFound := false
 	for i, line := range csvLines {
 		if i == 0 {
 			tokensMap = parseHeader(csvLines[0])
 			continue
 		}
 		tokens := strings.Split(line, ",")
 		if tokens[tokensMap["pxname"]] == poolName {
+			poolFound = true
 			if host := tokens[tokensMap["svname"]]; host != "BACKEND" && host != "FRONTEND" {
-				if status := tokens[tokensMap["status"]]; status == "UP" || status == "DOWN" {
+				status := tokens[tokensMap["status"]]
+				status = strings.Split(status, " ")[0]
+				if status == "UP" || status == "DOWN" {
 					hosts = append(hosts, host)
 				}
 			}
 		}
 	}
+	if !poolFound {
+		return hosts, fmt.Errorf("Haproxy CSV parsing error: did not find %+v pool", poolName)
+	}
 	return hosts, nil
 }
 
@@ -61,6 +76,11 @@ func ParseCsvHosts(csv string, poolName string) (hosts []string, err error) {
 // Read will read HAProxy URI and return with the CSV text
 func Read(host string, port int) (csv string, err error) {
 	haproxyUrl := fmt.Sprintf("http://%s:%d/;csv;norefresh", host, port)
+
+	if cachedCSV, found := csvCache.Get(haproxyUrl); found {
+		return cachedCSV.(string), nil
+	}
+
 	resp, err := http.Get(haproxyUrl)
 	if resp != nil {
 		defer resp.Body.Close()
@@ -72,5 +92,8 @@ func Read(host string, port int) (csv string, err error) {
 	if err != nil {
 		return "", err
 	}
-	return string(body), nil
+
+	csv = string(body)
+	csvCache.Set(haproxyUrl, csv, cache.DefaultExpiration)
+	return csv, nil
 }

go/mysql/mysql_throttle_metric.go

@@ -36,15 +36,16 @@ func ReadThrottleMetric(probe *Probe) (mySQLThrottleMetric *MySQLThrottleMetric)
 	mySQLThrottleMetric.Key = probe.Key
 
 	defer func(metric *MySQLThrottleMetric, started time.Time) {
-		metrics.GetOrRegisterTimer("probes.latency", nil).Update(time.Since(started))
-		metrics.GetOrRegisterCounter("probes.total", nil).Inc(1)
-		if metric.Err != nil {
-			metrics.GetOrRegisterCounter("probes.error", nil).Inc(1)
-		}
+		go func() {
+			metrics.GetOrRegisterTimer("probes.latency", nil).Update(time.Since(started))
+			metrics.GetOrRegisterCounter("probes.total", nil).Inc(1)
+			if metric.Err != nil {
+				metrics.GetOrRegisterCounter("probes.error", nil).Inc(1)
+			}
+		}()
 	}(mySQLThrottleMetric, started)
 
 	dbUri := probe.GetDBUri("information_schema")
-
 	db, fromCache, err := sqlutils.GetDB(dbUri)
 
 	if err != nil {

go/mysql/probe.go

@@ -28,7 +28,7 @@ type Probes map[InstanceKey](*Probe)
 type ClusterProbes struct {
 	ClusterName      string
 	IgnoreHostsCount int
-	Probes           *Probes
+	InstanceProbes   *Probes
 }
 
 func NewProbes() *Probes {

go/throttle/throttler.go

@@ -211,26 +211,26 @@ func (throttler *Throttler) refreshMySQLInventory() error {
 				clusterProbes := &mysql.ClusterProbes{
 					ClusterName:      clusterName,
 					IgnoreHostsCount: clusterSettings.IgnoreHostsCount,
-					Probes:           mysql.NewProbes(),
+					InstanceProbes:   mysql.NewProbes(),
 				}
 				for _, host := range hosts {
 					key := mysql.InstanceKey{Hostname: host, Port: clusterSettings.Port}
-					addInstanceKey(&key, clusterSettings, clusterProbes.Probes)
+					addInstanceKey(&key, clusterSettings, clusterProbes.InstanceProbes)
 				}
 				throttler.mysqlClusterProbesChan <- clusterProbes
 				return nil
 			}
 			if !clusterSettings.StaticHostsSettings.IsEmpty() {
 				clusterProbes := &mysql.ClusterProbes{
-					ClusterName: clusterName,
-					Probes:      mysql.NewProbes(),
+					ClusterName:    clusterName,
+					InstanceProbes: mysql.NewProbes(),
 				}
 				for _, host := range clusterSettings.StaticHostsSettings.Hosts {
 					key, err := mysql.ParseInstanceKey(host, clusterSettings.Port)
 					if err != nil {
 						return log.Errore(err)
 					}
-					addInstanceKey(key, clusterSettings, clusterProbes.Probes)
+					addInstanceKey(key, clusterSettings, clusterProbes.InstanceProbes)
 				}
 				throttler.mysqlClusterProbesChan <- clusterProbes
 				return nil
@@ -244,7 +244,7 @@ func (throttler *Throttler) refreshMySQLInventory() error {
 // synchronous update of inventory
 func (throttler *Throttler) updateMySQLClusterProbes(clusterProbes *mysql.ClusterProbes) error {
 	log.Debugf("onMySQLClusterProbes: %s", clusterProbes.ClusterName)
-	throttler.mysqlInventory.ClustersProbes[clusterProbes.ClusterName] = clusterProbes.Probes
+	throttler.mysqlInventory.ClustersProbes[clusterProbes.ClusterName] = clusterProbes.InstanceProbes
 	throttler.mysqlInventory.IgnoreHostsCount[clusterProbes.ClusterName] = clusterProbes.IgnoreHostsCount
 	return nil
 }