Add rendezvous hash for selecting subset of nodes

Select a user-defined number of pods via
rendezvous hash. This is important for larger
clusters, where the metric cardinality explosion
is too much for a single prometheus to handle.

Signed-off-by: stuart nelson <stuartnelson3@gmail.com>

Gopkg.lock

@@ -33,6 +33,14 @@
   pruneopts = "UT"
   revision = "3a771d992973f24aa725d07868b467d1ddfceafb"
 
+[[projects]]
+  digest = "1:998cf998358a303ac2430c386ba3fd3398477d6013153d3c6e11432765cc9ae6"
+  name = "github.com/cespare/xxhash"
+  packages = ["."]
+  pruneopts = "UT"
+  revision = "3b82fb7d186719faeedd0c2864f868c74fbf79a1"
+  version = "v2.0.0"
+
 [[projects]]
   digest = "1:6f82cacd0af5921e99bf3f46748705239b36489464f4529a1589bc895764fb18"
   name = "github.com/docker/go-units"
@@ -355,6 +363,14 @@
   revision = "583c0c0531f06d5278b7d917446061adc344b5cd"
   version = "v1.0.1"
 
+[[projects]]
+  digest = "1:2731ee5852c26a3585f70d1a6cff8422383b98bad405eb6611cd4a2334be1df6"
+  name = "github.com/stuartnelson3/go-rendezvous"
+  packages = ["."]
+  pruneopts = "UT"
+  revision = "16dc0292e5f420d89a7dc4319775a1e5e461dc01"
+  version = "v0.2.0"
+
 [[projects]]
   digest = "1:3f3a05ae0b95893d90b9b3b5afdb79a9b3d96e4e36e099d841ae602e4aca0da8"
   name = "golang.org/x/crypto"
@@ -625,6 +641,7 @@
   analyzer-name = "dep"
   analyzer-version = 1
   input-imports = [
+    "github.com/cespare/xxhash",
     "github.com/go-openapi/errors",
     "github.com/go-openapi/loads",
     "github.com/go-openapi/runtime",
@@ -639,6 +656,7 @@
     "github.com/jessevdk/go-flags",
     "github.com/prometheus/client_golang/prometheus",
     "github.com/prometheus/client_golang/prometheus/promhttp",
+    "github.com/stuartnelson3/go-rendezvous",
     "golang.org/x/image/font",
     "golang.org/x/image/font/basicfont",
     "golang.org/x/image/math/fixed",

Gopkg.toml

@@ -73,6 +73,10 @@
   name = "k8s.io/client-go"
   version = "9.0.0"
 
+[[constraint]]
+  name = "github.com/stuartnelson3/go-rendezvous"
+  version = "0.2.0"
+
 [prune]
   go-tests = true
   unused-packages = true

README.md

@@ -73,7 +73,7 @@ Building from source code consists of compiling the binary and building a [Docke
 
 ```sh
 # step 0: check out the code into your $GOPATH
-go get github.com/bloomberg/goldpinger/cmd/goldpinger 
+go get github.com/bloomberg/goldpinger/cmd/goldpinger
 cd $GOPATH/src/github.com/bloomberg/goldpinger
 
 # step 1: download the dependencies via dep ensure
@@ -103,7 +103,7 @@ namespace="docker.io/myhandle/" make push
 ### Example YAML
 
 
-Here's an example of what you can do (using the in-cluster authentication to `Kubernetes` apiserver). 
+Here's an example of what you can do (using the in-cluster authentication to `Kubernetes` apiserver).
 
 :warning: Replace `docker.io/mynamespace-replaceme/goldpinger:1.0.0` with the actual tag you built.
 
@@ -138,6 +138,11 @@ spec:
               valueFrom:
                 fieldRef:
                   fieldPath: spec.nodeName
+            # podIP is used to select a randomized subset of nodes to ping.
+            - name: POD_IP
+              valueFrom:
+                fieldRef:
+                  fieldPath: status.podIP
           image: "docker.io/mynamespace-replaceme/goldpinger:1.0.0"
           ports:
             - containerPort: 80
@@ -191,7 +196,7 @@ You can click on various nodes to gray out the clutter and see more information.
 
 ### API
 
-The API exposed is via a well-defined [`Swagger` spec](./swagger.yml). 
+The API exposed is via a well-defined [`Swagger` spec](./swagger.yml).
 
 The spec is used to generate both the server and the client of `Goldpinger`. If you make changes, you can re-generate them using [go-swagger](https://github.com/go-swagger/go-swagger) via [`make swagger`](./Makefile)
 

cmd/goldpinger/main.go

@@ -96,8 +96,16 @@ func main() {
 		goldpinger.GoldpingerConfig.Port = server.Port
 	}
 
-	server.ConfigureAPI()
-	goldpinger.StartUpdater()
+	if goldpinger.GoldpingerConfig.PodIP == "" {
+		log.Println("PodIP not set: pinging all pods")
+	}
+	if goldpinger.GoldpingerConfig.PingNumber == 0 {
+		log.Println("--ping-number set to 0: pinging all pods")
+	}
+	podSelecter := goldpinger.NewPodSelecter(goldpinger.GoldpingerConfig.PingNumber, goldpinger.GoldpingerConfig.PodIP, goldpinger.GetAllPods)
+
+	restapi.Configure(api, podSelecter)
+	goldpinger.StartUpdater(podSelecter)
 
 	log.Println("All good, starting serving the API")
 

extras/example-serviceaccounts.yml

@@ -1,4 +1,3 @@
-
 ---
 apiVersion: v1
 kind: ServiceAccount
@@ -38,6 +37,11 @@ spec:
               valueFrom:
                 fieldRef:
                   fieldPath: spec.nodeName
+            # podIP is used to select a randomized subset of nodes to ping.
+            - name: POD_IP
+              valueFrom:
+                fieldRef:
+                  fieldPath: status.podIP
           image: "docker.io/mynamespace-replaceme/goldpinger:1.1.0"
           ports:
             - containerPort: 80

extras/example-with-kubeconfig.yaml

@@ -43,6 +43,11 @@ spec:
               valueFrom:
                 fieldRef:
                   fieldPath: spec.nodeName
+            # podIP is used to select randomized subset of nodes to ping.
+            - name: POD_IP
+              valueFrom:
+                fieldRef:
+                  fieldPath: status.podIP
           image: "docker.io/mynamespace-replaceme/goldpinger:1.1.0"
           ports:
             - containerPort: 80

pkg/goldpinger/client.go

@@ -27,14 +27,14 @@ import (
 
 // CheckNeighbours queries the kubernetes API server for all other goldpinger pods
 // then calls Ping() on each one
-func CheckNeighbours() models.CheckResults {
-	return PingAllPods(GetAllPods())
+func CheckNeighbours(ps *PodSelecter) models.CheckResults {
+	return PingAllPods(ps.SelectPods())
 }
 
 // CheckNeighboursNeighbours queries the kubernetes API server for all other goldpinger
 // pods then calls Check() on each one
-func CheckNeighboursNeighbours() *models.CheckAllResults {
-	return CheckAllPods(GetAllPods())
+func CheckNeighboursNeighbours(ps *PodSelecter) *models.CheckAllResults {
+	return CheckAllPods(ps.SelectPods())
 }
 
 type PingAllPodsResult struct {

pkg/goldpinger/config.go

@@ -24,6 +24,8 @@ var GoldpingerConfig = struct {
 	KubeConfigPath   string `long:"kubeconfig" description:"Path to kubeconfig file" env:"KUBECONFIG"`
 	RefreshInterval  int    `long:"refresh-interval" description:"If > 0, will create a thread and collect stats every n seconds" env:"REFRESH_INTERVAL" default:"30"`
 	Hostname         string `long:"hostname" description:"Hostname to use" env:"HOSTNAME"`
+	PodIP            string `long:"pod-ip" description:"Pod IP to use" env:"POD_IP"`
+	PingNumber       uint   `long:"ping-number" description:"Number of peers to ping. A value of 0 indicates all peers should be pinged." default:"0"`
 	Port             int    `long:"client-port-override" description:"(for testing) use this port when calling other instances" env:"CLIENT_PORT_OVERRIDE"`
 	UseHostIP        bool   `long:"use-host-ip" description:"When making the calls, use host ip (defaults to pod ip)" env:"USE_HOST_IP"`
 	LabelSelector    string `long:"label-selector" description:"label selector to use to discover goldpinger pods in the cluster" env:"LABEL_SELECTOR" default:"app=goldpinger"`

pkg/goldpinger/pod_selecter.go

@@ -0,0 +1,61 @@
+// Copyright 2018 Bloomberg Finance L.P.
+//
+// Licensed 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.
+
+package goldpinger
+
+import (
+	"github.com/cespare/xxhash"
+	rendezvous "github.com/stuartnelson3/go-rendezvous"
+)
+
+// PodSelecter selects the result of getPods() down to count instances
+// according to a rendezvous hash.
+type PodSelecter struct {
+	count   uint
+	podIP   string
+	getPods func() map[string]string
+}
+
+// NewPodSelecter creates a new PodSelecter struct.
+func NewPodSelecter(count uint, podIP string, getPods func() map[string]string) *PodSelecter {
+	if podIP == "" {
+		// If podIP is blank, then we can't use the rendezvous hash to
+		// assign the IP correctly. Setting count=0 will force all pods
+		// to be pinged.
+		count = 0
+	}
+	return &PodSelecter{
+		count:   count,
+		podIP:   podIP,
+		getPods: getPods,
+	}
+}
+
+// SelectPods returns a map of pods filtered according to its configuration.
+func (p *PodSelecter) SelectPods() map[string]string {
+	allPods := p.getPods()
+	if p.count == 0 || p.count >= uint(len(allPods)) {
+		return allPods
+	}
+	rzv := rendezvous.New([]string{}, rendezvous.Hasher(xxhash.Sum64String))
+	for podIP := range allPods {
+		rzv.Add(podIP)
+	}
+	matches := rzv.LookupN(p.podIP, p.count)
+	toPing := make(map[string]string)
+	for _, podIP := range matches {
+		toPing[podIP] = allPods[podIP]
+	}
+	return toPing
+}

pkg/goldpinger/updater.go

@@ -20,16 +20,16 @@ import (
 	"time"
 )
 
-func StartUpdater() {
-
+func StartUpdater(ps *PodSelecter) {
 	if GoldpingerConfig.RefreshInterval <= 0 {
 		log.Println("Not creating updater, period is 0")
 		return
 	}
+
 	// start the updater
 	go func() {
 		for {
-			results := PingAllPods(GetAllPods())
+			results := PingAllPods(ps.SelectPods())
 			var troublemakers []string
 			for podIP, value := range results {
 				if *value.OK != true {

pkg/restapi/configure_goldpinger.go

@@ -47,6 +47,7 @@ func configureFlags(api *operations.GoldpingerAPI) {
 
 func configureAPI(api *operations.GoldpingerAPI) http.Handler {
 	// configure the api here
+	ps := &goldpinger.PodSelecter{}
 	api.ServeError = errors.ServeError
 
 	api.JSONConsumer = runtime.JSONConsumer()
@@ -61,13 +62,13 @@ func configureAPI(api *operations.GoldpingerAPI) http.Handler {
 	api.CheckServicePodsHandler = operations.CheckServicePodsHandlerFunc(
 		func(params operations.CheckServicePodsParams) middleware.Responder {
 			goldpinger.CountCall("received", "check")
-			return operations.NewCheckServicePodsOK().WithPayload(goldpinger.CheckNeighbours())
+			return operations.NewCheckServicePodsOK().WithPayload(goldpinger.CheckNeighbours(ps))
 		})
 
 	api.CheckAllPodsHandler = operations.CheckAllPodsHandlerFunc(
 		func(params operations.CheckAllPodsParams) middleware.Responder {
 			goldpinger.CountCall("received", "check_all")
-			return operations.NewCheckAllPodsOK().WithPayload(goldpinger.CheckNeighboursNeighbours())
+			return operations.NewCheckAllPodsOK().WithPayload(goldpinger.CheckNeighboursNeighbours(ps))
 		})
 
 	api.HealthzHandler = operations.HealthzHandlerFunc(
@@ -137,3 +138,44 @@ func prometheusMetricsMiddleware(next http.Handler) http.Handler {
 func setupGlobalMiddleware(handler http.Handler) http.Handler {
 	return prometheusMetricsMiddleware(fileServerMiddleware(handler))
 }
+
+// Configure sets the handlers on GoldpingerAPI.
+func Configure(api *operations.GoldpingerAPI, ps *goldpinger.PodSelecter) http.Handler {
+	api.ServeError = errors.ServeError
+
+	api.JSONConsumer = runtime.JSONConsumer()
+	api.JSONProducer = runtime.JSONProducer()
+
+	api.PingHandler = operations.PingHandlerFunc(
+		func(params operations.PingParams) middleware.Responder {
+			goldpinger.CountCall("received", "ping")
+			return operations.NewPingOK().WithPayload(goldpinger.GetStats())
+		})
+
+	api.CheckServicePodsHandler = operations.CheckServicePodsHandlerFunc(
+		func(params operations.CheckServicePodsParams) middleware.Responder {
+			goldpinger.CountCall("received", "check")
+			return operations.NewCheckServicePodsOK().WithPayload(goldpinger.CheckNeighbours(ps))
+		})
+
+	api.CheckAllPodsHandler = operations.CheckAllPodsHandlerFunc(
+		func(params operations.CheckAllPodsParams) middleware.Responder {
+			goldpinger.CountCall("received", "check_all")
+			return operations.NewCheckAllPodsOK().WithPayload(goldpinger.CheckNeighboursNeighbours(ps))
+		})
+
+	api.HealthzHandler = operations.HealthzHandlerFunc(
+		func(params operations.HealthzParams) middleware.Responder {
+			goldpinger.CountCall("received", "healthz")
+			healthResult := goldpinger.HealthCheck()
+			if *healthResult.OK {
+				return operations.NewHealthzOK().WithPayload(healthResult)
+			} else {
+				return operations.NewHealthzServiceUnavailable().WithPayload(healthResult)
+			}
+		})
+
+	api.ServerShutdown = func() {}
+
+	return setupGlobalMiddleware(api.Serve(setupMiddlewares))
+}