diff --git a/pkg/goldpinger/client.go b/pkg/goldpinger/client.go index 2c00c25..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 @@ -45,8 +57,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 { @@ -74,112 +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 - var channelResult PingAllPodsResult - channelResult.podName = pod.Name - channelResult.hostIPv4.UnmarshalText([]byte(pod.HostIP)) - channelResult.podIPv4.UnmarshalText([]byte(pod.PodIP)) - - OK := false - var responseTime int64 - 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, - 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: channelResult.podIPv4, - HostIP: channelResult.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: channelResult.podIPv4, - HostIP: channelResult.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/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"` diff --git a/pkg/goldpinger/pinger.go b/pkg/goldpinger/pinger.go new file mode 100644 index 0000000..e83cd1e --- /dev/null +++ b/pkg/goldpinger/pinger.go @@ -0,0 +1,176 @@ +// 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 ( + "context" + "time" + + "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/go-openapi/strfmt" + "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 { + pod *GoldpingerPod + client *apiclient.Goldpinger + timeout time.Duration + histogram prometheus.Observer + hostIPv4 strfmt.IPv4 + podIPv4 strfmt.IPv4 + 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(pod *GoldpingerPod, resultsChan chan<- PingAllPodsResult) *Pinger { + p := Pinger{ + pod: pod, + timeout: time.Duration(GoldpingerConfig.PingTimeoutMs) * time.Millisecond, + resultsChan: resultsChan, + stopChan: make(chan struct{}), + + histogram: goldpingerResponseTimePeersHistogram.WithLabelValues( + GoldpingerConfig.Hostname, + "ping", + 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 host/pod IPv4 + p.hostIPv4.UnmarshalText([]byte(pod.HostIP)) + p.podIPv4.UnmarshalText([]byte(pod.PodIP)) + + 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{ + PingTime: strfmt.DateTime(time.Now()), + 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() + + ctx, cancel := context.WithTimeout(context.Background(), p.timeout) + defer cancel() + + params := operations.NewPingParamsWithContext(ctx) + 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.resultsChan <- PingAllPodsResult{ + podName: p.pod.Name, + podResult: models.PodResult{ + PingTime: strfmt.DateTime(start), + 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.resultsChan <- PingAllPodsResult{ + podName: p.pod.Name, + podResult: models.PodResult{ + PingTime: strfmt.DateTime(start), + 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") + } +} + +// PingContinuously continuously pings the given pod with a delay between +// `period` and `period + jitterFactor * period` +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 c9546d6..8329f81 100644 --- a/pkg/goldpinger/updater.go +++ b/pkg/goldpinger/updater.go @@ -15,39 +15,179 @@ package goldpinger import ( - "context" - "fmt" + "sync" "time" "go.uber.org/zap" + + "github.com/bloomberg/goldpinger/v3/pkg/models" ) +// checkResults holds the latest results of checking the pods +var checkResults = models.CheckResults{PodResults: make(map[string]models.PodResult)} + +// checkResultsMux controls concurrent access to checkResults +var checkResultsMux = sync.Mutex{} + +// 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) +} + +// 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 + + 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 for new pods", + zap.Int("numNewPods", len(newPods)), + zap.Duration("refreshPeriod", refreshPeriod), + zap.Duration("waitPeriod", waitBetweenPods), + zap.Float64("JitterFactor", GoldpingerConfig.JitterFactor), + ) + + 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 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++ + } + } + 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) { + 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() + } + } +} + 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 := SelectPods() - // start the updater - go func() { - for { - ctx, cancel := context.WithTimeout(context.Background(), updateInterval) - - 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) - } - }() + // Create a channel for the results + resultsChan := make(chan PingAllPodsResult, len(pods)) + go updatePingers(resultsChan) + go collectResults(resultsChan) } 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