-
Notifications
You must be signed in to change notification settings - Fork 0
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Add health checker implementation (#938)
* Add health checker implementation *Populate the healthchecker Ip *Ping the healthchecker IP to check the remote gateway status *Update the gateway satus if the ping fails Fixes: submariner-io/submariner#821 Signed-off-by: Aswin Surayanarayanan <[email protected]> Co-authored-by: Thomas Pantelis <[email protected]>
- Loading branch information
1 parent
c99f0e8
commit 4287bb3
Showing
12 changed files
with
467 additions
and
25 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,132 @@ | ||
| package healthchecker | ||
|
|
||
| import ( | ||
| "sync" | ||
|
|
||
| "github.com/submariner-io/admiral/pkg/log" | ||
| "github.com/submariner-io/admiral/pkg/watcher" | ||
| submarinerv1 "github.com/submariner-io/submariner/pkg/apis/submariner.io/v1" | ||
| "k8s.io/apimachinery/pkg/runtime" | ||
| "k8s.io/klog" | ||
| ) | ||
|
|
||
| type LatencyInfo struct { | ||
| ConnectionError string | ||
| Spec *submarinerv1.LatencySpec | ||
| } | ||
|
|
||
| type Interface interface { | ||
| Start(stopCh <-chan struct{}) error | ||
|
|
||
| GetLatencyInfo(endpoint *submarinerv1.EndpointSpec) *LatencyInfo | ||
| } | ||
|
|
||
| type controller struct { | ||
| endpointWatcher watcher.Interface | ||
| pingers sync.Map | ||
| clusterID string | ||
| } | ||
|
|
||
| func New(config *watcher.Config, endpointNameSpace, clusterID string) (Interface, error) { | ||
| controller := &controller{ | ||
| clusterID: clusterID, | ||
| } | ||
| config.ResourceConfigs = []watcher.ResourceConfig{ | ||
| { | ||
| Name: "HealthChecker Endpoint Controller", | ||
| ResourceType: &submarinerv1.Endpoint{}, | ||
| Handler: watcher.EventHandlerFuncs{ | ||
| OnCreateFunc: controller.endpointCreatedorUpdated, | ||
| OnUpdateFunc: controller.endpointCreatedorUpdated, | ||
| OnDeleteFunc: controller.endpointDeleted, | ||
| }, | ||
| SourceNamespace: endpointNameSpace, | ||
| }, | ||
| } | ||
|
|
||
| endpointWatcher, err := watcher.New(config) | ||
|
|
||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| controller.endpointWatcher = endpointWatcher | ||
|
|
||
| return controller, nil | ||
| } | ||
|
|
||
| func (h *controller) GetLatencyInfo(endpoint *submarinerv1.EndpointSpec) *LatencyInfo { | ||
| if obj, found := h.pingers.Load(endpoint.CableName); found { | ||
| pinger := obj.(*pingerInfo) | ||
|
|
||
| return &LatencyInfo{ | ||
| ConnectionError: pinger.failureMsg, | ||
| Spec: &submarinerv1.LatencySpec{ | ||
| LastRTT: pinger.statistics.lastRtt, | ||
| MinRTT: pinger.statistics.minRtt, | ||
| AverageRTT: pinger.statistics.mean, | ||
| MaxRTT: pinger.statistics.maxRtt, | ||
| StdDevRTT: pinger.statistics.stdDev, | ||
| }, | ||
| } | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func (h *controller) Start(stopCh <-chan struct{}) error { | ||
| if err := h.endpointWatcher.Start(stopCh); err != nil { | ||
| return err | ||
| } | ||
|
|
||
| return nil | ||
| } | ||
|
|
||
| func (h *controller) endpointCreatedorUpdated(obj runtime.Object) bool { | ||
| klog.V(log.TRACE).Infof("Endpoint created: %#v", obj) | ||
| endpointCreated := obj.(*submarinerv1.Endpoint) | ||
| if endpointCreated.Spec.ClusterID == h.clusterID { | ||
| return false | ||
| } | ||
|
|
||
| if endpointCreated.Spec.HealthCheckIP == "" || endpointCreated.Spec.CableName == "" { | ||
| klog.Infof("HealthCheckIP (%q) and/or CableName (%q) for Endpoint %q empty - will not monitor endpoint health", | ||
| endpointCreated.Spec.HealthCheckIP, endpointCreated.Spec.CableName, endpointCreated.Name) | ||
| return false | ||
| } | ||
|
|
||
| if obj, found := h.pingers.Load(endpointCreated.Spec.CableName); found { | ||
| pinger := obj.(*pingerInfo) | ||
| if pinger.healthCheckIP == endpointCreated.Spec.HealthCheckIP { | ||
| return false | ||
| } | ||
|
|
||
| klog.V(log.DEBUG).Infof("HealthChecker is already running for %q - stopping", endpointCreated.Name) | ||
| pinger.stop() | ||
| h.pingers.Delete(endpointCreated.Spec.CableName) | ||
| } | ||
|
|
||
| klog.V(log.TRACE).Infof("Starting Pinger for CableName: %q, with HealthCheckIP: %q", | ||
| endpointCreated.Spec.CableName, endpointCreated.Spec.HealthCheckIP) | ||
|
|
||
| pinger := newPinger(endpointCreated.Spec.HealthCheckIP) | ||
| h.pingers.Store(endpointCreated.Spec.CableName, pinger) | ||
| pinger.start() | ||
|
|
||
| return false | ||
| } | ||
|
|
||
| func (h *controller) endpointDeleted(obj runtime.Object) bool { | ||
| endpointDeleted := obj.(*submarinerv1.Endpoint) | ||
| if endpointDeleted.Spec.CableName == "" { | ||
| return false | ||
| } | ||
|
|
||
| if obj, found := h.pingers.Load(endpointDeleted.Spec.CableName); found { | ||
| pinger := obj.(*pingerInfo) | ||
| pinger.stop() | ||
| h.pingers.Delete(endpointDeleted.Spec.CableName) | ||
| } | ||
|
|
||
| return false | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,13 @@ | ||
| package healthchecker | ||
|
|
||
| import ( | ||
| "testing" | ||
|
|
||
| . "github.com/onsi/ginkgo" | ||
| . "github.com/onsi/gomega" | ||
| ) | ||
|
|
||
| func TestHealthChecker(t *testing.T) { | ||
| RegisterFailHandler(Fail) | ||
| RunSpecs(t, "Health Checker Suite") | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,81 @@ | ||
| package healthchecker | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "time" | ||
|
|
||
| "github.com/go-ping/ping" | ||
| "k8s.io/klog" | ||
| ) | ||
|
|
||
| var waitTime time.Duration = 15 * time.Second | ||
| var timeout = 3 * time.Second | ||
|
|
||
| // The RTT will be stored and will be used to calculate the statistics until | ||
| // the size is reached. Once the size is reached the array will be reset and | ||
| // the last elements will be added to the array for statistics. | ||
| var size uint64 = 1000 | ||
|
|
||
| type pingerInfo struct { | ||
| healthCheckIP string | ||
| statistics statistics | ||
| failureMsg string | ||
| stopCh chan struct{} | ||
| } | ||
|
|
||
| func newPinger(healthCheckIP string) *pingerInfo { | ||
| return &pingerInfo{ | ||
| healthCheckIP: healthCheckIP, | ||
| statistics: statistics{ | ||
| size: size, | ||
| previousRtts: make([]uint64, size), | ||
| }, | ||
| stopCh: make(chan struct{}), | ||
| } | ||
| } | ||
|
|
||
| func (p *pingerInfo) start() { | ||
| go func() { | ||
| for { | ||
| select { | ||
| case <-p.stopCh: | ||
| return | ||
| case <-time.After(waitTime): | ||
| p.sendPing() | ||
| } | ||
| } | ||
| }() | ||
| klog.Infof("CableEngine HealthChecker started pinger for IP %q", p.healthCheckIP) | ||
| } | ||
|
|
||
| func (p *pingerInfo) stop() { | ||
| close(p.stopCh) | ||
| } | ||
|
|
||
| func (p *pingerInfo) sendPing() { | ||
| pinger, err := ping.NewPinger(p.healthCheckIP) | ||
| if err != nil { | ||
| klog.Errorf("Error creating pinger for IP %q: %v", p.healthCheckIP, err) | ||
| return | ||
| } | ||
|
|
||
| pinger.SetPrivileged(true) | ||
| pinger.RecordRtts = false | ||
| // After 3 seconds stop waiting. | ||
| pinger.Timeout = timeout | ||
|
|
||
| pinger.OnRecv = func(packet *ping.Packet) { | ||
| p.failureMsg = "" | ||
| p.statistics.update(uint64(packet.Rtt.Nanoseconds())) | ||
| } | ||
|
|
||
| pinger.OnFinish = func(stats *ping.Statistics) { | ||
| // Since we are setting a timeout and not a count, it will be an endless ping. | ||
| // If the timeout is reached with no successful packets, onFinish will be called and it is a failed ping. | ||
| p.failureMsg = fmt.Sprintf("Failed to successfully ping the remote endpoint IP %q", p.healthCheckIP) | ||
| } | ||
| err = pinger.Run() | ||
| if err != nil { | ||
| klog.Errorf("Error running ping for the remote endpoint IP %q: %v", p.healthCheckIP, err) | ||
| } | ||
| } |
Oops, something went wrong.