From d0dfd3e49328fca83b4268cbbd35294076cac22d Mon Sep 17 00:00:00 2001 From: Sachin Kamboj Date: Fri, 3 Apr 2020 07:46:17 -0400 Subject: [PATCH 1/8] Add code to continuously ping the pods and send the results over a channel Signed-off-by: Sachin Kamboj --- pkg/goldpinger/pinger.go | 86 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 86 insertions(+) create mode 100644 pkg/goldpinger/pinger.go diff --git a/pkg/goldpinger/pinger.go b/pkg/goldpinger/pinger.go new file mode 100644 index 0000000..5ce613a --- /dev/null +++ b/pkg/goldpinger/pinger.go @@ -0,0 +1,86 @@ +package goldpinger + +import ( + "log" + "time" + + apiclient "github.com/bloomberg/goldpinger/pkg/client" + "github.com/bloomberg/goldpinger/pkg/models" + "github.com/prometheus/client_golang/prometheus" + "k8s.io/apimachinery/pkg/util/wait" +) + +// Pinger contains all the info needed by a goroutine to continuously ping a pod +type Pinger struct { + podIP string + hostIP string + client *apiclient.Goldpinger + timer *prometheus.Timer + histogram prometheus.Observer + result PingAllPodsResult + resultsChan chan<- PingAllPodsResult + stopChan chan struct{} +} + +// NewPinger constructs and returns a Pinger object responsible for pinging a single +// goldpinger pod +func NewPinger(podIP string, hostIP string, resultsChan chan<- PingAllPodsResult) *Pinger { + p := Pinger{ + podIP: podIP, + hostIP: hostIP, + resultsChan: resultsChan, + stopChan: make(chan struct{}), + + histogram: goldpingerResponseTimePeersHistogram.WithLabelValues( + GoldpingerConfig.Hostname, + "ping", + hostIP, + podIP, + ), + } + + // Initialize the result + p.result.hostIPv4.UnmarshalText([]byte(hostIP)) + p.result.podIP = podIP + + // Get a client for pinging the given pod + // On error, create a static pod result that does nothing + client, err := getClient(pickPodHostIP(podIP, hostIP)) + if err == nil { + p.client = client + } else { + OK := false + p.client = nil + p.result.podResult = models.PodResult{HostIP: p.result.hostIPv4, OK: &OK, Error: err.Error(), StatusCode: 500, ResponseTimeMs: 0} + p.result.podIP = hostIP + } + return &p +} + +// Ping makes a single ping request to the given pod +func (p *Pinger) Ping() { + CountCall("made", "ping") + start := time.Now() + + resp, err := p.client.Operations.Ping(nil) + responseTime := time.Since(start) + responseTimeMs := responseTime.Nanoseconds() / int64(time.Millisecond) + p.histogram.Observe(responseTime.Seconds()) + + OK := (err == nil) + if OK { + p.result.podResult = models.PodResult{HostIP: p.result.hostIPv4, OK: &OK, Response: resp.Payload, StatusCode: 200, ResponseTimeMs: responseTimeMs} + log.Printf("Success pinging pod: %s, host: %s, resp: %+v, response time: %+v", p.podIP, p.hostIP, resp.Payload, responseTime) + } else { + p.result.podResult = models.PodResult{HostIP: p.result.hostIPv4, OK: &OK, Error: err.Error(), StatusCode: 504, ResponseTimeMs: responseTimeMs} + log.Printf("Error pinging pod: %s, host: %s, err: %+v, response time: %+v", p.podIP, p.hostIP, err, responseTime) + CountError("ping") + } + p.resultsChan <- p.result +} + +// PingContinuously continuously pings the given pod with a delay between +// `period` and `period + jitterFactor * period` +func (p *Pinger) PingContinuously(period time.Duration, jitterFactor float64) { + wait.JitterUntil(p.Ping, period, jitterFactor, false, p.stopChan) +} From 0690ac21a25f221cb74375e346483589f4d7e6e6 Mon Sep 17 00:00:00 2001 From: Sachin Kamboj Date: Fri, 3 Apr 2020 07:47:28 -0400 Subject: [PATCH 2/8] Command line options for adding a jitter-factor Signed-off-by: Sachin Kamboj --- pkg/goldpinger/config.go | 1 + 1 file changed, 1 insertion(+) diff --git a/pkg/goldpinger/config.go b/pkg/goldpinger/config.go index db92f17..042a171 100644 --- a/pkg/goldpinger/config.go +++ b/pkg/goldpinger/config.go @@ -23,6 +23,7 @@ var GoldpingerConfig = struct { StaticFilePath string `long:"static-file-path" description:"Folder for serving static files" env:"STATIC_FILE_PATH"` 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"` + JitterFactor float64 `long:"jitter-factor" description:"The amount of jitter to add while pinging clients" env:"JITTER_FACTOR" default:"0.05"` Hostname string `long:"hostname" description:"Hostname to use" env:"HOSTNAME"` PodIP string `long:"pod-ip" description:"Pod IP to use" env:"POD_IP"` PodName string `long:"pod-name" description:"The name of this pod - used to select --ping-number of pods using rendezvous hashing" env:"POD_NAME"` From 8a40aee927b6619ebd71385ddba10c901afd8cd9 Mon Sep 17 00:00:00 2001 From: Sachin Kamboj Date: Fri, 3 Apr 2020 07:47:59 -0400 Subject: [PATCH 3/8] Have the updater continuously ping pods and collate the results Signed-off-by: Sachin Kamboj --- pkg/goldpinger/updater.go | 108 +++++++++++++++++++++++++++++++------- 1 file changed, 88 insertions(+), 20 deletions(-) diff --git a/pkg/goldpinger/updater.go b/pkg/goldpinger/updater.go index c9546d6..676a953 100644 --- a/pkg/goldpinger/updater.go +++ b/pkg/goldpinger/updater.go @@ -18,36 +18,104 @@ import ( "context" "fmt" "time" + "sync" "go.uber.org/zap" + + "github.com/bloomberg/goldpinger/pkg/models" + "github.com/go-openapi/strfmt" ) +// resultsMux controls access to the results from multiple goroutines +var resultsMux sync.Mutex + +// getPingers creates a new set of pingers for the given pods +// Each pinger is responsible for pinging a single pod and returns +// the results on the results channel +func getPingers(pods map[string]string, resultsChan chan<- PingAllPodsResult) map[string]*Pinger { + pingers := map[string]*Pinger{} + + for podIP, hostIP := range pods { + pingers[podIP] = NewPinger(podIP, hostIP, resultsChan) + } + return pingers +} + +// startPingers starts `n` goroutines to continuously ping all the given pods, one goroutine per pod +// It staggers the start of all the go-routines to prevent a thundering herd +func startPingers(pingers map[string]*Pinger) { + refreshPeriod := time.Duration(GoldpingerConfig.RefreshInterval) * time.Second + waitBetweenPods := refreshPeriod / time.Duration(len(pingers)) + + log.Printf("Refresh Period: %+v Wait Period: %+v Jitter Factor: %+v", refreshPeriod, waitBetweenPods, GoldpingerConfig.JitterFactor) + + for _, p := range pingers { + go p.PingContinuously(refreshPeriod, GoldpingerConfig.JitterFactor) + time.Sleep(waitBetweenPods) + } +} + +// collectResults simply reads results from the results channel and saves them in a map +func collectResults(resultsChan <-chan PingAllPodsResult) *models.CheckResults { + results := models.CheckResults{} + results.PodResults = make(map[string]models.PodResult) + go func() { + for response := range resultsChan { + var podIPv4 strfmt.IPv4 + podIPv4.UnmarshalText([]byte(response.podIP)) + + // results.PodResults map will be read by processResults() + // to count the number of healthy and unhealthy nodes + // Since concurrent access to maps isn't safe from multiple + // goroutines, lock the mutex before update + resultsMux.Lock() + results.PodResults[response.podIP] = response.podResult + resultsMux.Unlock() + } + }() + return &results +} + +// processResults goes through all the entries in the results channel and counts +// the number of health and unhealth nodes. It just reports the correct number +func processResults(results *models.CheckResults) { + for { + var troublemakers []string + var counterHealthy, counterUnhealthy float64 + + resultsMux.Lock() + for podIP, value := range results.PodResults { + if *value.OK != true { + counterUnhealthy++ + troublemakers = append(troublemakers, fmt.Sprintf("%s (%s)", podIP, value.HostIP.String())) + } else { + counterHealthy++ + } + } + resultsMux.Unlock() + + CountHealthyUnhealthyNodes(counterHealthy, counterUnhealthy) + if len(troublemakers) > 0 { + log.Println("Updater ran into trouble with these peers: ", troublemakers) + } + time.Sleep(time.Duration(GoldpingerConfig.RefreshInterval) * time.Second) + } +} + func StartUpdater() { if GoldpingerConfig.RefreshInterval <= 0 { zap.L().Info("Not creating updater, refresh interval is negative", zap.Int("RefreshInterval", GoldpingerConfig.RefreshInterval)) return } - updateInterval := time.Duration(GoldpingerConfig.RefreshInterval) * time.Second + pods := GoldpingerConfig.PodSelecter.SelectPods() + zap.S().Infof("Got Pods: %+v", pods) - // start the updater - go func() { - for { - ctx, cancel := context.WithTimeout(context.Background(), updateInterval) + // Create a channel for the results + resultsChan := make(chan PingAllPodsResult, len(pods)) + pingers := getPingers(pods, resultsChan) - results := PingAllPods(ctx, SelectPods()) - var troublemakers []string - for podIP, value := range results.PodResults { - if *value.OK != true { - troublemakers = append(troublemakers, fmt.Sprintf("%s (%s)", podIP, value.HostIP.String())) - } - } - if len(troublemakers) > 0 { - zap.L().Warn("Updater ran into trouble with these peers", zap.Strings("troublemakers", troublemakers)) - } - - cancel() - time.Sleep(updateInterval) - } - }() + startPingers(pingers) + results := collectResults(resultsChan) + go processResults(results) } From 40f57b1a4e55218113b0db138cc7ca7bfcc4bc3a Mon Sep 17 00:00:00 2001 From: Sachin Kamboj Date: Fri, 3 Apr 2020 08:56:30 -0400 Subject: [PATCH 4/8] Get rid of the lock and keep a running count of healthy/unhealthy nodes Signed-off-by: Sachin Kamboj --- pkg/goldpinger/pinger.go | 67 +++++++++++++++------- pkg/goldpinger/updater.go | 113 +++++++++++++++++++------------------- 2 files changed, 104 insertions(+), 76 deletions(-) diff --git a/pkg/goldpinger/pinger.go b/pkg/goldpinger/pinger.go index 5ce613a..c743265 100644 --- a/pkg/goldpinger/pinger.go +++ b/pkg/goldpinger/pinger.go @@ -1,58 +1,67 @@ package goldpinger import ( - "log" + "context" "time" - apiclient "github.com/bloomberg/goldpinger/pkg/client" - "github.com/bloomberg/goldpinger/pkg/models" + "go.uber.org/zap" + + apiclient "github.com/bloomberg/goldpinger/v3/pkg/client" + "github.com/bloomberg/goldpinger/v3/pkg/client/operations" + "github.com/bloomberg/goldpinger/v3/pkg/models" "github.com/prometheus/client_golang/prometheus" "k8s.io/apimachinery/pkg/util/wait" ) // Pinger contains all the info needed by a goroutine to continuously ping a pod type Pinger struct { - podIP string - hostIP string + pod *GoldpingerPod client *apiclient.Goldpinger - timer *prometheus.Timer + timeout time.Duration histogram prometheus.Observer result PingAllPodsResult resultsChan chan<- PingAllPodsResult stopChan chan struct{} + logger *zap.Logger } // NewPinger constructs and returns a Pinger object responsible for pinging a single // goldpinger pod -func NewPinger(podIP string, hostIP string, resultsChan chan<- PingAllPodsResult) *Pinger { +func NewPinger(pod *GoldpingerPod, resultsChan chan<- PingAllPodsResult) *Pinger { p := Pinger{ - podIP: podIP, - hostIP: hostIP, + pod: pod, + timeout: time.Duration(GoldpingerConfig.PingTimeoutMs) * time.Millisecond, resultsChan: resultsChan, stopChan: make(chan struct{}), histogram: goldpingerResponseTimePeersHistogram.WithLabelValues( GoldpingerConfig.Hostname, "ping", - hostIP, - podIP, + pod.HostIP, + pod.PodIP, + ), + + logger: zap.L().With( + zap.String("op", "pinger"), + zap.String("name", pod.Name), + zap.String("hostIP", pod.HostIP), + zap.String("podIP", pod.PodIP), ), } // Initialize the result - p.result.hostIPv4.UnmarshalText([]byte(hostIP)) - p.result.podIP = podIP + p.result.hostIPv4.UnmarshalText([]byte(pod.HostIP)) + p.result.podIPv4.UnmarshalText([]byte(pod.PodIP)) // Get a client for pinging the given pod // On error, create a static pod result that does nothing - client, err := getClient(pickPodHostIP(podIP, hostIP)) + client, err := getClient(pickPodHostIP(pod.PodIP, pod.HostIP)) if err == nil { p.client = client } else { OK := false p.client = nil p.result.podResult = models.PodResult{HostIP: p.result.hostIPv4, OK: &OK, Error: err.Error(), StatusCode: 500, ResponseTimeMs: 0} - p.result.podIP = hostIP } return &p } @@ -62,18 +71,36 @@ func (p *Pinger) Ping() { CountCall("made", "ping") start := time.Now() - resp, err := p.client.Operations.Ping(nil) + ctx, cancel := context.WithTimeout(context.Background(), p.timeout) + defer cancel() + + params := operations.NewPingParamsWithContext(ctx) + resp, err := p.client.Operations.Ping(params) responseTime := time.Since(start) responseTimeMs := responseTime.Nanoseconds() / int64(time.Millisecond) p.histogram.Observe(responseTime.Seconds()) OK := (err == nil) if OK { - p.result.podResult = models.PodResult{HostIP: p.result.hostIPv4, OK: &OK, Response: resp.Payload, StatusCode: 200, ResponseTimeMs: responseTimeMs} - log.Printf("Success pinging pod: %s, host: %s, resp: %+v, response time: %+v", p.podIP, p.hostIP, resp.Payload, responseTime) + p.result.podResult = models.PodResult{ + PodIP: p.result.podIPv4, + HostIP: p.result.hostIPv4, + OK: &OK, + Response: resp.Payload, + StatusCode: 200, + ResponseTimeMs: responseTimeMs, + } + p.logger.Debug("Success pinging pod", zap.Duration("responseTime", responseTime)) } else { - p.result.podResult = models.PodResult{HostIP: p.result.hostIPv4, OK: &OK, Error: err.Error(), StatusCode: 504, ResponseTimeMs: responseTimeMs} - log.Printf("Error pinging pod: %s, host: %s, err: %+v, response time: %+v", p.podIP, p.hostIP, err, responseTime) + p.result.podResult = models.PodResult{ + PodIP: p.result.podIPv4, + HostIP: p.result.hostIPv4, + OK: &OK, + Error: err.Error(), + StatusCode: 504, + ResponseTimeMs: responseTimeMs, + } + p.logger.Warn("Ping returned error", zap.Duration("responseTime", responseTime), zap.Error(err)) CountError("ping") } p.resultsChan <- p.result diff --git a/pkg/goldpinger/updater.go b/pkg/goldpinger/updater.go index 676a953..b12567b 100644 --- a/pkg/goldpinger/updater.go +++ b/pkg/goldpinger/updater.go @@ -15,39 +15,54 @@ package goldpinger import ( - "context" - "fmt" "time" - "sync" "go.uber.org/zap" - "github.com/bloomberg/goldpinger/pkg/models" - "github.com/go-openapi/strfmt" + "github.com/bloomberg/goldpinger/v3/pkg/models" ) -// resultsMux controls access to the results from multiple goroutines -var resultsMux sync.Mutex +// checkResults holds the latest results of checking the pods +var checkResults models.CheckResults + +// counterHealthy is the number of healthy pods +var counterHealthy float64 // getPingers creates a new set of pingers for the given pods // Each pinger is responsible for pinging a single pod and returns // the results on the results channel -func getPingers(pods map[string]string, resultsChan chan<- PingAllPodsResult) map[string]*Pinger { +func getPingers(pods map[string]*GoldpingerPod, resultsChan chan<- PingAllPodsResult) map[string]*Pinger { pingers := map[string]*Pinger{} - for podIP, hostIP := range pods { - pingers[podIP] = NewPinger(podIP, hostIP, resultsChan) + for podName, pod := range pods { + pingers[podName] = NewPinger(pod, resultsChan) } return pingers } +// initCheckResults initializes the check results, which will be updated continuously +// as the results come in +func initCheckResults(pingers map[string]*Pinger) { + checkResults = models.CheckResults{} + checkResults.PodResults = make(map[string]models.PodResult) + for podName, pinger := range pingers { + checkResults.PodResults[podName] = pinger.result.podResult + } + counterHealthy = 0 +} + // startPingers starts `n` goroutines to continuously ping all the given pods, one goroutine per pod // It staggers the start of all the go-routines to prevent a thundering herd func startPingers(pingers map[string]*Pinger) { refreshPeriod := time.Duration(GoldpingerConfig.RefreshInterval) * time.Second waitBetweenPods := refreshPeriod / time.Duration(len(pingers)) - log.Printf("Refresh Period: %+v Wait Period: %+v Jitter Factor: %+v", refreshPeriod, waitBetweenPods, GoldpingerConfig.JitterFactor) + zap.L().Info( + "Starting Pingers", + zap.Duration("refreshPeriod", refreshPeriod), + zap.Duration("waitPeriod", waitBetweenPods), + zap.Float64("JitterFactor", GoldpingerConfig.JitterFactor), + ) for _, p := range pingers { go p.PingContinuously(refreshPeriod, GoldpingerConfig.JitterFactor) @@ -55,51 +70,38 @@ func startPingers(pingers map[string]*Pinger) { } } -// collectResults simply reads results from the results channel and saves them in a map -func collectResults(resultsChan <-chan PingAllPodsResult) *models.CheckResults { - results := models.CheckResults{} - results.PodResults = make(map[string]models.PodResult) - go func() { - for response := range resultsChan { - var podIPv4 strfmt.IPv4 - podIPv4.UnmarshalText([]byte(response.podIP)) +// updateCounters updates the value of health and unhealthy nodes as the results come in +func updateCounters(podName string, result *models.PodResult) { + // Get the previous value of ok + old := checkResults.PodResults[podName] + oldOk := (old.OK != nil && *old.OK) - // results.PodResults map will be read by processResults() - // to count the number of healthy and unhealthy nodes - // Since concurrent access to maps isn't safe from multiple - // goroutines, lock the mutex before update - resultsMux.Lock() - results.PodResults[response.podIP] = response.podResult - resultsMux.Unlock() - } - }() - return &results + // Check if the value of ok has changed + // If not, do nothing + if oldOk == *result.OK { + return + } + + if *result.OK { + // The value was previously false and just became true + // Increment the counter + counterHealthy++ + } else { + // The value was previously true and just became false + counterHealthy-- + } + CountHealthyUnhealthyNodes(counterHealthy, float64(len(checkResults.PodResults))-counterHealthy) } -// processResults goes through all the entries in the results channel and counts -// the number of health and unhealth nodes. It just reports the correct number -func processResults(results *models.CheckResults) { - for { - var troublemakers []string - var counterHealthy, counterUnhealthy float64 - - resultsMux.Lock() - for podIP, value := range results.PodResults { - if *value.OK != true { - counterUnhealthy++ - troublemakers = append(troublemakers, fmt.Sprintf("%s (%s)", podIP, value.HostIP.String())) - } else { - counterHealthy++ - } +// collectResults simply reads results from the results channel and saves them in a map +func collectResults(resultsChan <-chan PingAllPodsResult) { + go func() { + for response := range resultsChan { + result := response.podResult + updateCounters(response.podName, &result) + checkResults.PodResults[response.podName] = result } - resultsMux.Unlock() - - CountHealthyUnhealthyNodes(counterHealthy, counterUnhealthy) - if len(troublemakers) > 0 { - log.Println("Updater ran into trouble with these peers: ", troublemakers) - } - time.Sleep(time.Duration(GoldpingerConfig.RefreshInterval) * time.Second) - } + }() } func StartUpdater() { @@ -108,14 +110,13 @@ func StartUpdater() { return } - pods := GoldpingerConfig.PodSelecter.SelectPods() + pods := SelectPods() zap.S().Infof("Got Pods: %+v", pods) // Create a channel for the results resultsChan := make(chan PingAllPodsResult, len(pods)) pingers := getPingers(pods, resultsChan) - + initCheckResults(pingers) startPingers(pingers) - results := collectResults(resultsChan) - go processResults(results) + collectResults(resultsChan) } From 9db241d67d9ff43b15caaac6a371ec4ab63c9b1b Mon Sep 17 00:00:00 2001 From: Sachin Kamboj Date: Tue, 7 Apr 2020 19:54:19 -0400 Subject: [PATCH 5/8] Update the set of pingers at regular intervals from the k8s API server Signed-off-by: Sachin Kamboj --- pkg/goldpinger/client.go | 31 +++---- pkg/goldpinger/pinger.go | 124 ++++++++++++++++++------- pkg/goldpinger/updater.go | 184 ++++++++++++++++++++++++++++---------- 3 files changed, 243 insertions(+), 96 deletions(-) diff --git a/pkg/goldpinger/client.go b/pkg/goldpinger/client.go index 2c00c25..12be2b2 100644 --- a/pkg/goldpinger/client.go +++ b/pkg/goldpinger/client.go @@ -45,8 +45,7 @@ func CheckNeighboursNeighbours(ctx context.Context) *models.CheckAllResults { type PingAllPodsResult struct { podName string podResult models.PodResult - hostIPv4 strfmt.IPv4 - podIPv4 strfmt.IPv4 + deleted bool } func pickPodHostIP(podIP, hostIP string) string { @@ -99,20 +98,22 @@ func PingAllPods(pingAllCtx context.Context, pods map[string]*GoldpingerPod) *mo start := time.Now() // setup - var channelResult PingAllPodsResult - channelResult.podName = pod.Name - channelResult.hostIPv4.UnmarshalText([]byte(pod.HostIP)) - channelResult.podIPv4.UnmarshalText([]byte(pod.PodIP)) - + channelResult := PingAllPodsResult{podName: pod.Name} OK := false - var responseTime int64 - client, err := getClient(pickPodHostIP(pod.PodIP, pod.HostIP)) + var responseTime int64 + var hostIPv4 strfmt.IPv4 + var podIPv4 strfmt.IPv4 + + hostIPv4.UnmarshalText([]byte(pod.HostIP)) + podIPv4.UnmarshalText([]byte(pod.PodIP)) + + client, err := getClient(pickPodHostIP(pod.PodIP, pod.HostIP)) if err != nil { logger.Warn("Couldn't get a client for Ping", zap.Error(err)) channelResult.podResult = models.PodResult{ - PodIP: channelResult.podIPv4, - HostIP: channelResult.hostIPv4, + PodIP: podIPv4, + HostIP: hostIPv4, OK: &OK, Error: err.Error(), StatusCode: 500, @@ -133,8 +134,8 @@ func PingAllPods(pingAllCtx context.Context, pods map[string]*GoldpingerPod) *mo if OK { logger.Debug("Pink Ok", zap.Int64("responseTime", responseTime)) channelResult.podResult = models.PodResult{ - PodIP: channelResult.podIPv4, - HostIP: channelResult.hostIPv4, + PodIP: podIPv4, + HostIP: hostIPv4, OK: &OK, Response: resp.Payload, StatusCode: 200, @@ -144,8 +145,8 @@ func PingAllPods(pingAllCtx context.Context, pods map[string]*GoldpingerPod) *mo } else { logger.Warn("Ping returned error", zap.Int64("responseTime", responseTime), zap.Error(err)) channelResult.podResult = models.PodResult{ - PodIP: channelResult.podIPv4, - HostIP: channelResult.hostIPv4, + PodIP: podIPv4, + HostIP: hostIPv4, OK: &OK, Error: err.Error(), StatusCode: 504, diff --git a/pkg/goldpinger/pinger.go b/pkg/goldpinger/pinger.go index c743265..6b06e3b 100644 --- a/pkg/goldpinger/pinger.go +++ b/pkg/goldpinger/pinger.go @@ -1,3 +1,17 @@ +// 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 ( @@ -9,6 +23,7 @@ import ( apiclient "github.com/bloomberg/goldpinger/v3/pkg/client" "github.com/bloomberg/goldpinger/v3/pkg/client/operations" "github.com/bloomberg/goldpinger/v3/pkg/models" + "github.com/go-openapi/strfmt" "github.com/prometheus/client_golang/prometheus" "k8s.io/apimachinery/pkg/util/wait" ) @@ -19,7 +34,8 @@ type Pinger struct { client *apiclient.Goldpinger timeout time.Duration histogram prometheus.Observer - result PingAllPodsResult + hostIPv4 strfmt.IPv4 + podIPv4 strfmt.IPv4 resultsChan chan<- PingAllPodsResult stopChan chan struct{} logger *zap.Logger @@ -49,25 +65,48 @@ func NewPinger(pod *GoldpingerPod, resultsChan chan<- PingAllPodsResult) *Pinger ), } - // Initialize the result - p.result.hostIPv4.UnmarshalText([]byte(pod.HostIP)) - p.result.podIPv4.UnmarshalText([]byte(pod.PodIP)) + // Initialize the host/pod IPv4 + p.hostIPv4.UnmarshalText([]byte(pod.HostIP)) + p.podIPv4.UnmarshalText([]byte(pod.PodIP)) - // Get a client for pinging the given pod - // On error, create a static pod result that does nothing - client, err := getClient(pickPodHostIP(pod.PodIP, pod.HostIP)) - if err == nil { - p.client = client - } else { - OK := false - p.client = nil - p.result.podResult = models.PodResult{HostIP: p.result.hostIPv4, OK: &OK, Error: err.Error(), StatusCode: 500, ResponseTimeMs: 0} - } return &p } +// getClient returns a client that can be used to ping the given pod +// On error, it returns a static result +func (p *Pinger) getClient() (*apiclient.Goldpinger, error) { + if p.client != nil { + return p.client, nil + } + + client, err := getClient(pickPodHostIP(p.pod.PodIP, p.pod.HostIP)) + if err != nil { + p.logger.Warn("Could not get client", zap.Error(err)) + OK := false + p.resultsChan <- PingAllPodsResult{ + podName: p.pod.Name, + podResult: models.PodResult{ + PodIP: p.podIPv4, + HostIP: p.hostIPv4, + OK: &OK, + Error: err.Error(), + StatusCode: 500, + ResponseTimeMs: 0, + }, + } + return nil, err + } + p.client = client + return p.client, nil +} + // Ping makes a single ping request to the given pod func (p *Pinger) Ping() { + client, err := p.getClient() + if err != nil { + return + } + CountCall("made", "ping") start := time.Now() @@ -75,39 +114,60 @@ func (p *Pinger) Ping() { defer cancel() params := operations.NewPingParamsWithContext(ctx) - resp, err := p.client.Operations.Ping(params) + resp, err := client.Operations.Ping(params) responseTime := time.Since(start) responseTimeMs := responseTime.Nanoseconds() / int64(time.Millisecond) p.histogram.Observe(responseTime.Seconds()) OK := (err == nil) if OK { - p.result.podResult = models.PodResult{ - PodIP: p.result.podIPv4, - HostIP: p.result.hostIPv4, - OK: &OK, - Response: resp.Payload, - StatusCode: 200, - ResponseTimeMs: responseTimeMs, + p.resultsChan <- PingAllPodsResult{ + podName: p.pod.Name, + podResult: models.PodResult{ + PodIP: p.podIPv4, + HostIP: p.hostIPv4, + OK: &OK, + Response: resp.Payload, + StatusCode: 200, + ResponseTimeMs: responseTimeMs, + }, } p.logger.Debug("Success pinging pod", zap.Duration("responseTime", responseTime)) } else { - p.result.podResult = models.PodResult{ - PodIP: p.result.podIPv4, - HostIP: p.result.hostIPv4, - OK: &OK, - Error: err.Error(), - StatusCode: 504, - ResponseTimeMs: responseTimeMs, + p.resultsChan <- PingAllPodsResult{ + podName: p.pod.Name, + podResult: models.PodResult{ + PodIP: p.podIPv4, + HostIP: p.hostIPv4, + OK: &OK, + Error: err.Error(), + StatusCode: 504, + ResponseTimeMs: responseTimeMs, + }, } p.logger.Warn("Ping returned error", zap.Duration("responseTime", responseTime), zap.Error(err)) CountError("ping") } - p.resultsChan <- p.result } // PingContinuously continuously pings the given pod with a delay between // `period` and `period + jitterFactor * period` -func (p *Pinger) PingContinuously(period time.Duration, jitterFactor float64) { - wait.JitterUntil(p.Ping, period, jitterFactor, false, p.stopChan) +func (p *Pinger) PingContinuously(initialWait time.Duration, period time.Duration, jitterFactor float64) { + p.logger.Info( + "Starting pinger", + zap.Duration("period", period), + zap.Duration("initialWait", initialWait), + zap.Float64("jitterFactor", jitterFactor), + ) + + timer := time.NewTimer(initialWait) + + select { + case <-timer.C: + wait.JitterUntil(p.Ping, period, jitterFactor, false, p.stopChan) + case <-p.stopChan: + // Do nothing + } + // We are done, send a message on the results channel to delete this + p.resultsChan <- PingAllPodsResult{podName: p.pod.Name, deleted: true} } diff --git a/pkg/goldpinger/updater.go b/pkg/goldpinger/updater.go index b12567b..813739a 100644 --- a/pkg/goldpinger/updater.go +++ b/pkg/goldpinger/updater.go @@ -15,6 +15,7 @@ package goldpinger import ( + "sync" "time" "go.uber.org/zap" @@ -23,71 +24,156 @@ import ( ) // checkResults holds the latest results of checking the pods -var checkResults models.CheckResults +var checkResults = models.CheckResults{PodResults: make(map[string]models.PodResult)} // counterHealthy is the number of healthy pods -var counterHealthy float64 +var counterHealthy = float64(0.0) -// getPingers creates a new set of pingers for the given pods -// Each pinger is responsible for pinging a single pod and returns -// the results on the results channel -func getPingers(pods map[string]*GoldpingerPod, resultsChan chan<- PingAllPodsResult) map[string]*Pinger { - pingers := map[string]*Pinger{} +// checkResultsMux controls concurrent access to checkResults +var checkResultsMux = sync.Mutex{} - for podName, pod := range pods { - pingers[podName] = NewPinger(pod, resultsChan) - } - return pingers +// exists checks whether there is an existing pinger for the given pod +// returns true if: +// - there is already a pinger with the same name +// - the pinger has the same podIP +// - the pinger has the same hostIP +func exists(existingPods map[string]*GoldpingerPod, new *GoldpingerPod) bool { + old, exists := existingPods[new.Name] + return exists && (old.PodIP == new.PodIP) && (old.HostIP == new.HostIP) } -// initCheckResults initializes the check results, which will be updated continuously -// as the results come in -func initCheckResults(pingers map[string]*Pinger) { - checkResults = models.CheckResults{} - checkResults.PodResults = make(map[string]models.PodResult) - for podName, pinger := range pingers { - checkResults.PodResults[podName] = pinger.result.podResult - } - counterHealthy = 0 -} - -// startPingers starts `n` goroutines to continuously ping all the given pods, one goroutine per pod -// It staggers the start of all the go-routines to prevent a thundering herd -func startPingers(pingers map[string]*Pinger) { +// updatePingers calls SelectPods() at regular intervals to get a new list of goldpinger pods to ping +// For each goldpinger pod, it then creates a pinger responsible for pinging it and returning the +// results on the result channel +func updatePingers(resultsChan chan<- PingAllPodsResult) { + // Important: This is the only goroutine that should have access to + // these maps since there is nothing controlling concurrent access + pingers := make(map[string]*Pinger) + existingPods := make(map[string]*GoldpingerPod) refreshPeriod := time.Duration(GoldpingerConfig.RefreshInterval) * time.Second - waitBetweenPods := refreshPeriod / time.Duration(len(pingers)) + + for { + // Initialize deletedPods to all existing pods, we will remove + // any pods that should still exist from this list after we are done + // NOTE: This is *NOT* a copy of existingPods just a new variable name + // to make the intention/code clear and cleaner + deletedPods := existingPods + + // New pods are brand new and haven't been seen before + newPods := make(map[string]*GoldpingerPod) + + latest := SelectPods() + for podName, pod := range latest { + if exists(existingPods, pod) { + // This pod continues to exist in the latest iteration of the update + // without any changes + // Delete it from the set of pods that we wish to delete + delete(deletedPods, podName) + } else { + // This pod is brand new and has never been seen before + // Add it to the list of newPods + newPods[podName] = pod + } + } + + // deletedPods now contains any pods that have either been deleted from the api-server + // *OR* weren't selected by our rendezvous hash + // *OR* had their host/pod IP changed. Remove those pingers + destroyPingers(pingers, deletedPods) + + // Next create pingers for new pods + createPingers(pingers, newPods, resultsChan, refreshPeriod) + + // Finally, just set existingPods to the latest and collect garbage + existingPods = latest + deletedPods = nil + newPods = nil + + // Wait the given time before pinging + time.Sleep(refreshPeriod) + } +} + +// createPingers allocates a new pinger object for each new goldpinger Pod that's been discovered +// It also: +// (a) initializes a result object in checkResults to store info on that pod +// (b) starts a new goroutines to continuously ping the given pod. +// Each new goroutine waits for a given time before starting the continuous ping +// to prevent a thundering herd +func createPingers(pingers map[string]*Pinger, newPods map[string]*GoldpingerPod, resultsChan chan<- PingAllPodsResult, refreshPeriod time.Duration) { + if len(newPods) == 0 { + // I have nothing to do + return + } + waitBetweenPods := refreshPeriod / time.Duration(len(newPods)) zap.L().Info( - "Starting Pingers", + "Starting pingers for new pods", + zap.Int("numNewPods", len(newPods)), zap.Duration("refreshPeriod", refreshPeriod), zap.Duration("waitPeriod", waitBetweenPods), zap.Float64("JitterFactor", GoldpingerConfig.JitterFactor), ) - for _, p := range pingers { - go p.PingContinuously(refreshPeriod, GoldpingerConfig.JitterFactor) - time.Sleep(waitBetweenPods) + initialWait := time.Duration(0) + for podName, pod := range newPods { + pinger := NewPinger(pod, resultsChan) + pingers[podName] = pinger + go pinger.PingContinuously(initialWait, refreshPeriod, GoldpingerConfig.JitterFactor) + initialWait += waitBetweenPods + } +} + +// destroyPingers takes a list of deleted pods and then for each pod in the list, it stops +// the goroutines that continuously pings that pod and then deletes the pod from the list of pingers +func destroyPingers(pingers map[string]*Pinger, deletedPods map[string]*GoldpingerPod) { + for podName, pod := range deletedPods { + zap.L().Info( + "Deleting pod from pingers", + zap.String("name", podName), + zap.String("podIP", pod.PodIP), + zap.String("hostIP", pod.HostIP), + ) + pinger := pingers[podName] + + // Close the channel to stop pinging + close(pinger.stopChan) + + // delete from pingers + delete(pingers, podName) } } // updateCounters updates the value of health and unhealthy nodes as the results come in func updateCounters(podName string, result *models.PodResult) { // Get the previous value of ok - old := checkResults.PodResults[podName] - oldOk := (old.OK != nil && *old.OK) - - // Check if the value of ok has changed - // If not, do nothing - if oldOk == *result.OK { - return - } - - if *result.OK { + old, oldExists := checkResults.PodResults[podName] + switch { + case result == nil: + // This pod was just deleted + // If the previous value seen was ok, decrement + // the counter + if oldExists && old.OK != nil && *old.OK { + counterHealthy-- + } + case !oldExists || old.OK == nil: + // If there is no previous response, this is an initialization step for this pod name + // The default is unhealthy, so: + // - if this pod is healthy, increment the count + // - if this pod is unhealthy, do not touch the count + if *result.OK { + counterHealthy++ + } + case *old.OK == *result.OK: + // The old value is equal to the new value + // Do nothing! + case *result.OK: // The value was previously false and just became true // Increment the counter counterHealthy++ - } else { + default: // The value was previously true and just became false + // Decrement the counter counterHealthy-- } CountHealthyUnhealthyNodes(counterHealthy, float64(len(checkResults.PodResults))-counterHealthy) @@ -95,13 +181,16 @@ func updateCounters(podName string, result *models.PodResult) { // collectResults simply reads results from the results channel and saves them in a map func collectResults(resultsChan <-chan PingAllPodsResult) { - go func() { - for response := range resultsChan { + for response := range resultsChan { + if response.deleted { + updateCounters(response.podName, nil) + delete(checkResults.PodResults, response.podName) + } else { result := response.podResult updateCounters(response.podName, &result) checkResults.PodResults[response.podName] = result } - }() + } } func StartUpdater() { @@ -111,12 +200,9 @@ func StartUpdater() { } pods := SelectPods() - zap.S().Infof("Got Pods: %+v", pods) // Create a channel for the results resultsChan := make(chan PingAllPodsResult, len(pods)) - pingers := getPingers(pods, resultsChan) - initCheckResults(pingers) - startPingers(pingers) - collectResults(resultsChan) + go updatePingers(resultsChan) + go collectResults(resultsChan) } From 2a78a9cec599a5d12aeabf6d20532317f0824172 Mon Sep 17 00:00:00 2001 From: Sachin Kamboj Date: Tue, 7 Apr 2020 20:49:01 -0400 Subject: [PATCH 6/8] Don't ping all pods on call, return existing data Signed-off-by: Sachin Kamboj --- pkg/goldpinger/client.go | 122 ++++---------------------------------- pkg/goldpinger/updater.go | 2 + 2 files changed, 15 insertions(+), 109 deletions(-) diff --git a/pkg/goldpinger/client.go b/pkg/goldpinger/client.go index 12be2b2..ddaf67f 100644 --- a/pkg/goldpinger/client.go +++ b/pkg/goldpinger/client.go @@ -33,7 +33,19 @@ import ( // CheckNeighbours queries the kubernetes API server for all other goldpinger pods // then calls Ping() on each one func CheckNeighbours(ctx context.Context) *models.CheckResults { - return PingAllPods(ctx, SelectPods()) + // Mux to prevent concurrent map address + checkResultsMux.Lock() + defer checkResultsMux.Unlock() + + final := models.CheckResults{} + final.PodResults = make(map[string]models.PodResult) + for podName, podResult := range checkResults.PodResults { + final.PodResults[podName] = podResult + } + if len(GoldpingerConfig.DnsHosts) > 0 { + final.DNSResults = *checkDNS() + } + return &final } // CheckNeighboursNeighbours queries the kubernetes API server for all other goldpinger @@ -73,114 +85,6 @@ func checkDNS() *models.DNSResults { return &results } -func PingAllPods(pingAllCtx context.Context, pods map[string]*GoldpingerPod) *models.CheckResults { - - result := models.CheckResults{} - - ch := make(chan PingAllPodsResult, len(pods)) - wg := sync.WaitGroup{} - wg.Add(len(pods)) - - for _, pod := range pods { - - go func(pod *GoldpingerPod) { - - logger := zap.L().With( - zap.String("op", "ping"), - zap.String("name", pod.Name), - zap.String("hostIP", pod.HostIP), - zap.String("podIP", pod.PodIP), - ) - - // metrics - CountCall("made", "ping") - timer := GetLabeledPeersCallsTimer("ping", pod.HostIP, pod.PodIP) - start := time.Now() - - // setup - channelResult := PingAllPodsResult{podName: pod.Name} - OK := false - - var responseTime int64 - var hostIPv4 strfmt.IPv4 - var podIPv4 strfmt.IPv4 - - hostIPv4.UnmarshalText([]byte(pod.HostIP)) - podIPv4.UnmarshalText([]byte(pod.PodIP)) - - client, err := getClient(pickPodHostIP(pod.PodIP, pod.HostIP)) - if err != nil { - logger.Warn("Couldn't get a client for Ping", zap.Error(err)) - channelResult.podResult = models.PodResult{ - PodIP: podIPv4, - HostIP: hostIPv4, - OK: &OK, - Error: err.Error(), - StatusCode: 500, - ResponseTimeMs: responseTime, - } - CountError("ping") - } else { - pingCtx, cancel := context.WithTimeout( - pingAllCtx, - time.Duration(GoldpingerConfig.PingTimeoutMs)*time.Millisecond, - ) - defer cancel() - - params := operations.NewPingParamsWithContext(pingCtx) - resp, err := client.Operations.Ping(params) - responseTime = time.Since(start).Nanoseconds() / int64(time.Millisecond) - OK = (err == nil) - if OK { - logger.Debug("Pink Ok", zap.Int64("responseTime", responseTime)) - channelResult.podResult = models.PodResult{ - PodIP: podIPv4, - HostIP: hostIPv4, - OK: &OK, - Response: resp.Payload, - StatusCode: 200, - ResponseTimeMs: responseTime, - } - timer.ObserveDuration() - } else { - logger.Warn("Ping returned error", zap.Int64("responseTime", responseTime), zap.Error(err)) - channelResult.podResult = models.PodResult{ - PodIP: podIPv4, - HostIP: hostIPv4, - OK: &OK, - Error: err.Error(), - StatusCode: 504, - ResponseTimeMs: responseTime, - } - CountError("ping") - } - } - - ch <- channelResult - wg.Done() - }(pod) - } - if len(GoldpingerConfig.DnsHosts) > 0 { - result.DNSResults = *checkDNS() - } - wg.Wait() - close(ch) - - counterHealthy, counterUnhealthy := 0.0, 0.0 - - result.PodResults = make(map[string]models.PodResult) - for response := range ch { - if *response.podResult.OK { - counterHealthy++ - } else { - counterUnhealthy++ - } - result.PodResults[response.podName] = response.podResult - } - CountHealthyUnhealthyNodes(counterHealthy, counterUnhealthy) - return &result -} - type CheckServicePodsResult struct { podName string checkAllPodResult models.CheckAllPodResult diff --git a/pkg/goldpinger/updater.go b/pkg/goldpinger/updater.go index 813739a..e43cebf 100644 --- a/pkg/goldpinger/updater.go +++ b/pkg/goldpinger/updater.go @@ -182,6 +182,7 @@ func updateCounters(podName string, result *models.PodResult) { // collectResults simply reads results from the results channel and saves them in a map func collectResults(resultsChan <-chan PingAllPodsResult) { for response := range resultsChan { + checkResultsMux.Lock() if response.deleted { updateCounters(response.podName, nil) delete(checkResults.PodResults, response.podName) @@ -190,6 +191,7 @@ func collectResults(resultsChan <-chan PingAllPodsResult) { updateCounters(response.podName, &result) checkResults.PodResults[response.podName] = result } + checkResultsMux.Unlock() } } From d68d35bbabbefc2746a2c3bc34542f0343eb945e Mon Sep 17 00:00:00 2001 From: Sachin Kamboj Date: Tue, 7 Apr 2020 21:04:43 -0400 Subject: [PATCH 7/8] Add a ping time that gives the last time a node was pinged Signed-off-by: Sachin Kamboj --- pkg/goldpinger/pinger.go | 3 +++ pkg/models/pod_result.go | 21 +++++++++++++++++++++ pkg/restapi/embedded_spec.go | 8 ++++++++ swagger.yml | 3 +++ 4 files changed, 35 insertions(+) diff --git a/pkg/goldpinger/pinger.go b/pkg/goldpinger/pinger.go index 6b06e3b..e83cd1e 100644 --- a/pkg/goldpinger/pinger.go +++ b/pkg/goldpinger/pinger.go @@ -86,6 +86,7 @@ func (p *Pinger) getClient() (*apiclient.Goldpinger, error) { p.resultsChan <- PingAllPodsResult{ podName: p.pod.Name, podResult: models.PodResult{ + PingTime: strfmt.DateTime(time.Now()), PodIP: p.podIPv4, HostIP: p.hostIPv4, OK: &OK, @@ -124,6 +125,7 @@ func (p *Pinger) Ping() { p.resultsChan <- PingAllPodsResult{ podName: p.pod.Name, podResult: models.PodResult{ + PingTime: strfmt.DateTime(start), PodIP: p.podIPv4, HostIP: p.hostIPv4, OK: &OK, @@ -137,6 +139,7 @@ func (p *Pinger) Ping() { p.resultsChan <- PingAllPodsResult{ podName: p.pod.Name, podResult: models.PodResult{ + PingTime: strfmt.DateTime(start), PodIP: p.podIPv4, HostIP: p.hostIPv4, OK: &OK, diff --git a/pkg/models/pod_result.go b/pkg/models/pod_result.go index ea5b553..589df02 100644 --- a/pkg/models/pod_result.go +++ b/pkg/models/pod_result.go @@ -24,6 +24,10 @@ type PodResult struct { // o k OK *bool `json:"OK,omitempty"` + // ping time + // Format: date-time + PingTime strfmt.DateTime `json:"PingTime,omitempty"` + // pod IP // Format: ipv4 PodIP strfmt.IPv4 `json:"PodIP,omitempty"` @@ -49,6 +53,10 @@ func (m *PodResult) Validate(formats strfmt.Registry) error { res = append(res, err) } + if err := m.validatePingTime(formats); err != nil { + res = append(res, err) + } + if err := m.validatePodIP(formats); err != nil { res = append(res, err) } @@ -76,6 +84,19 @@ func (m *PodResult) validateHostIP(formats strfmt.Registry) error { return nil } +func (m *PodResult) validatePingTime(formats strfmt.Registry) error { + + if swag.IsZero(m.PingTime) { // not required + return nil + } + + if err := validate.FormatOf("PingTime", "body", "date-time", m.PingTime.String(), formats); err != nil { + return err + } + + return nil +} + func (m *PodResult) validatePodIP(formats strfmt.Registry) error { if swag.IsZero(m.PodIP) { // not required diff --git a/pkg/restapi/embedded_spec.go b/pkg/restapi/embedded_spec.go index b37bc1a..bc2d282 100644 --- a/pkg/restapi/embedded_spec.go +++ b/pkg/restapi/embedded_spec.go @@ -259,6 +259,10 @@ func init() { "type": "boolean", "default": false }, + "PingTime": { + "type": "string", + "format": "date-time" + }, "PodIP": { "type": "string", "format": "ipv4" @@ -527,6 +531,10 @@ func init() { "type": "boolean", "default": false }, + "PingTime": { + "type": "string", + "format": "date-time" + }, "PodIP": { "type": "string", "format": "ipv4" diff --git a/swagger.yml b/swagger.yml index 9f8d955..fd1dfd3 100644 --- a/swagger.yml +++ b/swagger.yml @@ -37,6 +37,9 @@ definitions: OK: type: boolean default: false + PingTime: + format: date-time + type: string PodIP: type: string format: ipv4 From 7e60ee675aa688f24699fa5045d73bd53fddf4b8 Mon Sep 17 00:00:00 2001 From: Sachin Kamboj Date: Wed, 8 Apr 2020 07:56:04 -0400 Subject: [PATCH 8/8] Simplify updateCounters, don't try to maintain a running count Signed-off-by: Sachin Kamboj --- pkg/goldpinger/updater.go | 67 +++++++++++++++------------------------ 1 file changed, 25 insertions(+), 42 deletions(-) diff --git a/pkg/goldpinger/updater.go b/pkg/goldpinger/updater.go index e43cebf..8329f81 100644 --- a/pkg/goldpinger/updater.go +++ b/pkg/goldpinger/updater.go @@ -26,9 +26,6 @@ import ( // checkResults holds the latest results of checking the pods var checkResults = models.CheckResults{PodResults: make(map[string]models.PodResult)} -// counterHealthy is the number of healthy pods -var counterHealthy = float64(0.0) - // checkResultsMux controls concurrent access to checkResults var checkResultsMux = sync.Mutex{} @@ -144,54 +141,40 @@ func destroyPingers(pingers map[string]*Pinger, deletedPods map[string]*Goldping } } -// updateCounters updates the value of health and unhealthy nodes as the results come in -func updateCounters(podName string, result *models.PodResult) { - // Get the previous value of ok - old, oldExists := checkResults.PodResults[podName] - switch { - case result == nil: - // This pod was just deleted - // If the previous value seen was ok, decrement - // the counter - if oldExists && old.OK != nil && *old.OK { - counterHealthy-- - } - case !oldExists || old.OK == nil: - // If there is no previous response, this is an initialization step for this pod name - // The default is unhealthy, so: - // - if this pod is healthy, increment the count - // - if this pod is unhealthy, do not touch the count - if *result.OK { +// updateCounters updates the count of health and unhealthy nodes +func updateCounters() { + checkResultsMux.Lock() + defer checkResultsMux.Unlock() + + var counterHealthy float64 + for _, result := range checkResults.PodResults { + if result.OK != nil && *result.OK { counterHealthy++ } - case *old.OK == *result.OK: - // The old value is equal to the new value - // Do nothing! - case *result.OK: - // The value was previously false and just became true - // Increment the counter - counterHealthy++ - default: - // The value was previously true and just became false - // Decrement the counter - counterHealthy-- } CountHealthyUnhealthyNodes(counterHealthy, float64(len(checkResults.PodResults))-counterHealthy) } // collectResults simply reads results from the results channel and saves them in a map func collectResults(resultsChan <-chan PingAllPodsResult) { - for response := range resultsChan { - checkResultsMux.Lock() - if response.deleted { - updateCounters(response.podName, nil) - delete(checkResults.PodResults, response.podName) - } else { - result := response.podResult - updateCounters(response.podName, &result) - checkResults.PodResults[response.podName] = result + refreshPeriod := time.Duration(GoldpingerConfig.RefreshInterval) * time.Second + updateTicker := time.NewTicker(refreshPeriod) + for { + select { + case <-updateTicker.C: + // Every time our update ticker ticks, update the count of healthy/unhealthy nodes + updateCounters() + case response := <-resultsChan: + // On getting a ping response, if the pinger is not being deleted, + // simply save it for later + checkResultsMux.Lock() + if response.deleted { + delete(checkResults.PodResults, response.podName) + } else { + checkResults.PodResults[response.podName] = response.podResult + } + checkResultsMux.Unlock() } - checkResultsMux.Unlock() } }