Merge branch 'master' into asPercent

Author: stivenbb

api/cluster.go

@@ -7,17 +7,31 @@ import (
 	"net/http"
 	"strconv"
 	"sync"
+	"time"
 
 	"github.com/grafana/metrictank/api/middleware"
 	"github.com/grafana/metrictank/api/models"
 	"github.com/grafana/metrictank/api/response"
 	"github.com/grafana/metrictank/cluster"
+	"github.com/grafana/metrictank/stats"
 	"github.com/raintank/worldping-api/pkg/log"
 	"github.com/tinylib/msgp/msgp"
 )
 
 var NotFoundErr = errors.New("not found")
 
+var (
+
+	// metric api.cluster.speculative.attempts is how many peer queries resulted in speculation
+	speculativeAttempts = stats.NewCounter32("api.cluster.speculative.attempts")
+
+	// metric api.cluster.speculative.wins is how many peer queries were improved due to speculation
+	speculativeWins = stats.NewCounter32("api.cluster.speculative.wins")
+
+	// metric api.cluster.speculative.requests is how many speculative http requests made to peers
+	speculativeRequests = stats.NewCounter32("api.cluster.speculative.requests")
+)
+
 func (s *Server) explainPriority(ctx *middleware.Context) {
 	var data []interface{}
 	for _, p := range s.prioritySetters {
@@ -287,3 +301,114 @@ func (s *Server) peerQuery(ctx context.Context, data cluster.Traceable, name, pa
 
 	return result, nil
 }
+
+// peerQuerySpeculative takes a request and the path to request it on, then fans it out
+// across the cluster, except to the local peer. If any peer fails requests to
+// other peers are aborted. If enough peers have been heard from (based on
+// speculation-threshold configuration), and we are missing the others, try to
+// speculatively query each other member of each shard group.
+// ctx:          request context
+// data:         request to be submitted
+// name:         name to be used in logging & tracing
+// path:         path to request on
+func (s *Server) peerQuerySpeculative(ctx context.Context, data cluster.Traceable, name, path string) (map[string]PeerResponse, error) {
+	peerGroups, err := cluster.MembersForSpeculativeQuery()
+	if err != nil {
+		log.Error(3, "HTTP peerQuery unable to get peers, %s", err)
+		return nil, err
+	}
+	log.Debug("HTTP %s across %d instances", name, len(peerGroups)-1)
+
+	reqCtx, cancel := context.WithCancel(ctx)
+	defer cancel()
+
+	originalPeers := make(map[string]struct{}, len(peerGroups))
+	receivedResponses := make(map[int32]struct{}, len(peerGroups))
+
+	responses := make(chan struct {
+		shardGroup int32
+		data       PeerResponse
+		err        error
+	}, 1)
+
+	askPeer := func(shardGroup int32, peer cluster.Node) {
+		log.Debug("HTTP Render querying %s%s", peer.GetName(), path)
+		buf, err := peer.Post(reqCtx, name, path, data)
+
+		select {
+		case <-ctx.Done():
+			return
+		default:
+			// Not canceled, continue
+		}
+
+		if err != nil {
+			cancel()
+			log.Error(4, "HTTP Render error querying %s%s: %q", peer.GetName(), path, err)
+		}
+		responses <- struct {
+			shardGroup int32
+			data       PeerResponse
+			err        error
+		}{shardGroup, PeerResponse{peer, buf}, err}
+	}
+
+	for group, peers := range peerGroups {
+		peer := peers[0]
+		originalPeers[peer.GetName()] = struct{}{}
+		go askPeer(group, peer)
+	}
+
+	result := make(map[string]PeerResponse)
+
+	var ticker *time.Ticker
+	var tickChan <-chan time.Time
+	if speculationThreshold != 1 {
+		ticker = time.NewTicker(5 * time.Millisecond)
+		tickChan = ticker.C
+		defer ticker.Stop()
+	}
+
+	for len(receivedResponses) < len(peerGroups) {
+		select {
+		case resp := <-responses:
+			if _, ok := receivedResponses[resp.shardGroup]; ok {
+				// already received this response (possibly speculatively)
+				continue
+			}
+
+			if resp.err != nil {
+				return nil, err
+			}
+
+			result[resp.data.peer.GetName()] = resp.data
+			receivedResponses[resp.shardGroup] = struct{}{}
+			delete(originalPeers, resp.data.peer.GetName())
+
+		case <-tickChan:
+			// Check if it's time to speculate!
+			percentReceived := float64(len(receivedResponses)) / float64(len(peerGroups))
+			if percentReceived >= speculationThreshold {
+				// kick off speculative queries to other members now
+				ticker.Stop()
+				speculativeAttempts.Inc()
+				for shardGroup, peers := range peerGroups {
+					if _, ok := receivedResponses[shardGroup]; ok {
+						continue
+					}
+					eligiblePeers := peers[1:]
+					for _, peer := range eligiblePeers {
+						speculativeRequests.Inc()
+						go askPeer(shardGroup, peer)
+					}
+				}
+			}
+		}
+	}
+
+	if len(originalPeers) > 0 {
+		speculativeWins.Inc()
+	}
+
+	return result, nil
+}

api/config.go

@@ -29,6 +29,7 @@ var (
 
 	getTargetsConcurrency int
 	tagdbDefaultLimit     uint
+	speculationThreshold  float64
 
 	graphiteProxy *httputil.ReverseProxy
 	timeZone      *time.Location
@@ -50,6 +51,7 @@ func ConfigSetup() {
 	apiCfg.StringVar(&timeZoneStr, "time-zone", "local", "timezone for interpreting from/until values when needed, specified using [zoneinfo name](https://en.wikipedia.org/wiki/Tz_database#Names_of_time_zones) e.g. 'America/New_York', 'UTC' or 'local' to use local server timezone")
 	apiCfg.IntVar(&getTargetsConcurrency, "get-targets-concurrency", 20, "maximum number of concurrent threads for fetching data on the local node. Each thread handles a single series.")
 	apiCfg.UintVar(&tagdbDefaultLimit, "tagdb-default-limit", 100, "default limit for tagdb query results, can be overridden with query parameter \"limit\"")
+	apiCfg.Float64Var(&speculationThreshold, "speculation-threshold", 1, "ratio of peer responses after which speculation is used. Set to 1 to disable.")
 	globalconf.Register("http", apiCfg)
 }