7cd5477c3c
In current implementation of Consul, check alias cannot determine if a service exists or not. Because a service without any check is semantically considered as passing, so when no healthchecks are found for an agent, the check was considered as passing. But this make little sense as the current implementation does not make any difference between: * a non-existing service (passing) * a service without any check (passing as well) In order to make it work, we have to ensure that when a check did not find any healthcheck, the service does indeed exists. If it does not, lets consider the check as failing.
857 lines
21 KiB
Go
857 lines
21 KiB
Go
package checks
|
|
|
|
import (
|
|
"crypto/tls"
|
|
"fmt"
|
|
"io"
|
|
"net"
|
|
"net/http"
|
|
"os"
|
|
osexec "os/exec"
|
|
"strings"
|
|
"sync"
|
|
"syscall"
|
|
"time"
|
|
|
|
"github.com/hashicorp/consul/agent/structs"
|
|
"github.com/hashicorp/consul/sdk/testutil"
|
|
"github.com/hashicorp/go-hclog"
|
|
|
|
"github.com/armon/circbuf"
|
|
"github.com/hashicorp/consul/agent/exec"
|
|
"github.com/hashicorp/consul/api"
|
|
"github.com/hashicorp/consul/lib"
|
|
"github.com/hashicorp/go-cleanhttp"
|
|
)
|
|
|
|
const (
|
|
// MinInterval is the minimal interval between
|
|
// two checks. Do not allow for a interval below this value.
|
|
// Otherwise we risk fork bombing a system.
|
|
MinInterval = time.Second
|
|
|
|
// DefaultBufSize is the maximum size of the captured
|
|
// check output by default. Prevents an enormous buffer
|
|
// from being captured
|
|
DefaultBufSize = 4 * 1024 // 4KB
|
|
|
|
// UserAgent is the value of the User-Agent header
|
|
// for HTTP health checks.
|
|
UserAgent = "Consul Health Check"
|
|
)
|
|
|
|
// RPC is an interface that an RPC client must implement. This is a helper
|
|
// interface that is implemented by the agent delegate for checks that need
|
|
// to make RPC calls.
|
|
type RPC interface {
|
|
RPC(method string, args interface{}, reply interface{}) error
|
|
}
|
|
|
|
// CheckNotifier interface is used by the CheckMonitor
|
|
// to notify when a check has a status update. The update
|
|
// should take care to be idempotent.
|
|
type CheckNotifier interface {
|
|
UpdateCheck(checkID structs.CheckID, status, output string)
|
|
// ServiceExists return true if the given service does exists
|
|
ServiceExists(serviceID structs.ServiceID) bool
|
|
}
|
|
|
|
// CheckMonitor is used to periodically invoke a script to
|
|
// determine the health of a given check. It is compatible with
|
|
// nagios plugins and expects the output in the same format.
|
|
// Supports failures_before_critical and success_before_passing.
|
|
type CheckMonitor struct {
|
|
Notify CheckNotifier
|
|
CheckID structs.CheckID
|
|
ServiceID structs.ServiceID
|
|
Script string
|
|
ScriptArgs []string
|
|
Interval time.Duration
|
|
Timeout time.Duration
|
|
Logger hclog.Logger
|
|
OutputMaxSize int
|
|
StatusHandler *StatusHandler
|
|
|
|
stop bool
|
|
stopCh chan struct{}
|
|
stopLock sync.Mutex
|
|
}
|
|
|
|
// Start is used to start a check monitor.
|
|
// Monitor runs until stop is called
|
|
func (c *CheckMonitor) Start() {
|
|
c.stopLock.Lock()
|
|
defer c.stopLock.Unlock()
|
|
c.stop = false
|
|
c.stopCh = make(chan struct{})
|
|
go c.run()
|
|
}
|
|
|
|
// Stop is used to stop a check monitor.
|
|
func (c *CheckMonitor) Stop() {
|
|
c.stopLock.Lock()
|
|
defer c.stopLock.Unlock()
|
|
if !c.stop {
|
|
c.stop = true
|
|
close(c.stopCh)
|
|
}
|
|
}
|
|
|
|
// run is invoked by a goroutine to run until Stop() is called
|
|
func (c *CheckMonitor) run() {
|
|
// Get the randomized initial pause time
|
|
initialPauseTime := lib.RandomStagger(c.Interval)
|
|
next := time.After(initialPauseTime)
|
|
for {
|
|
select {
|
|
case <-next:
|
|
c.check()
|
|
next = time.After(c.Interval)
|
|
case <-c.stopCh:
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// check is invoked periodically to perform the script check
|
|
func (c *CheckMonitor) check() {
|
|
// Create the command
|
|
var cmd *osexec.Cmd
|
|
var err error
|
|
if len(c.ScriptArgs) > 0 {
|
|
cmd, err = exec.Subprocess(c.ScriptArgs)
|
|
} else {
|
|
cmd, err = exec.Script(c.Script)
|
|
}
|
|
if err != nil {
|
|
c.Logger.Error("Check failed to setup",
|
|
"check", c.CheckID.String(),
|
|
"error", err,
|
|
)
|
|
c.Notify.UpdateCheck(c.CheckID, api.HealthCritical, err.Error())
|
|
return
|
|
}
|
|
|
|
// Collect the output
|
|
output, _ := circbuf.NewBuffer(int64(c.OutputMaxSize))
|
|
cmd.Stdout = output
|
|
cmd.Stderr = output
|
|
exec.SetSysProcAttr(cmd)
|
|
|
|
truncateAndLogOutput := func() string {
|
|
outputStr := string(output.Bytes())
|
|
if output.TotalWritten() > output.Size() {
|
|
outputStr = fmt.Sprintf("Captured %d of %d bytes\n...\n%s",
|
|
output.Size(), output.TotalWritten(), outputStr)
|
|
}
|
|
c.Logger.Trace("Check output",
|
|
"check", c.CheckID.String(),
|
|
"output", outputStr,
|
|
)
|
|
return outputStr
|
|
}
|
|
|
|
// Start the check
|
|
if err := cmd.Start(); err != nil {
|
|
c.Logger.Error("Check failed to invoke",
|
|
"check", c.CheckID.String(),
|
|
"error", err,
|
|
)
|
|
c.Notify.UpdateCheck(c.CheckID, api.HealthCritical, err.Error())
|
|
return
|
|
}
|
|
|
|
// Wait for the check to complete
|
|
waitCh := make(chan error, 1)
|
|
go func() {
|
|
waitCh <- cmd.Wait()
|
|
}()
|
|
|
|
timeout := 30 * time.Second
|
|
if c.Timeout > 0 {
|
|
timeout = c.Timeout
|
|
}
|
|
select {
|
|
case <-time.After(timeout):
|
|
if err := exec.KillCommandSubtree(cmd); err != nil {
|
|
c.Logger.Warn("Check failed to kill after timeout",
|
|
"check", c.CheckID.String(),
|
|
"error", err,
|
|
)
|
|
}
|
|
|
|
msg := fmt.Sprintf("Timed out (%s) running check", timeout.String())
|
|
c.Logger.Warn("Timed out running check",
|
|
"check", c.CheckID.String(),
|
|
"timeout", timeout.String(),
|
|
)
|
|
|
|
outputStr := truncateAndLogOutput()
|
|
if len(outputStr) > 0 {
|
|
msg += "\n\n" + outputStr
|
|
}
|
|
c.Notify.UpdateCheck(c.CheckID, api.HealthCritical, msg)
|
|
|
|
// Now wait for the process to exit so we never start another
|
|
// instance concurrently.
|
|
<-waitCh
|
|
return
|
|
|
|
case err = <-waitCh:
|
|
// The process returned before the timeout, proceed normally
|
|
}
|
|
|
|
// Check if the check passed
|
|
outputStr := truncateAndLogOutput()
|
|
if err == nil {
|
|
c.StatusHandler.updateCheck(c.CheckID, api.HealthPassing, outputStr)
|
|
return
|
|
}
|
|
|
|
// If the exit code is 1, set check as warning
|
|
exitErr, ok := err.(*osexec.ExitError)
|
|
if ok {
|
|
if status, ok := exitErr.Sys().(syscall.WaitStatus); ok {
|
|
code := status.ExitStatus()
|
|
if code == 1 {
|
|
c.StatusHandler.updateCheck(c.CheckID, api.HealthWarning, outputStr)
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// Set the health as critical
|
|
c.StatusHandler.updateCheck(c.CheckID, api.HealthCritical, outputStr)
|
|
}
|
|
|
|
// CheckTTL is used to apply a TTL to check status,
|
|
// and enables clients to set the status of a check
|
|
// but upon the TTL expiring, the check status is
|
|
// automatically set to critical.
|
|
type CheckTTL struct {
|
|
Notify CheckNotifier
|
|
CheckID structs.CheckID
|
|
ServiceID structs.ServiceID
|
|
TTL time.Duration
|
|
Logger hclog.Logger
|
|
|
|
timer *time.Timer
|
|
|
|
lastOutput string
|
|
lastOutputLock sync.RWMutex
|
|
|
|
stop bool
|
|
stopCh chan struct{}
|
|
stopLock sync.Mutex
|
|
|
|
OutputMaxSize int
|
|
}
|
|
|
|
// Start is used to start a check ttl, runs until Stop()
|
|
func (c *CheckTTL) Start() {
|
|
c.stopLock.Lock()
|
|
defer c.stopLock.Unlock()
|
|
if c.OutputMaxSize < 1 {
|
|
c.OutputMaxSize = DefaultBufSize
|
|
}
|
|
c.stop = false
|
|
c.stopCh = make(chan struct{})
|
|
c.timer = time.NewTimer(c.TTL)
|
|
go c.run()
|
|
}
|
|
|
|
// Stop is used to stop a check ttl.
|
|
func (c *CheckTTL) Stop() {
|
|
c.stopLock.Lock()
|
|
defer c.stopLock.Unlock()
|
|
if !c.stop {
|
|
c.timer.Stop()
|
|
c.stop = true
|
|
close(c.stopCh)
|
|
}
|
|
}
|
|
|
|
// run is used to handle TTL expiration and to update the check status
|
|
func (c *CheckTTL) run() {
|
|
for {
|
|
select {
|
|
case <-c.timer.C:
|
|
c.Logger.Warn("Check missed TTL, is now critical",
|
|
"check", c.CheckID.String(),
|
|
)
|
|
c.Notify.UpdateCheck(c.CheckID, api.HealthCritical, c.getExpiredOutput())
|
|
|
|
case <-c.stopCh:
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// getExpiredOutput formats the output for the case when the TTL is expired.
|
|
func (c *CheckTTL) getExpiredOutput() string {
|
|
c.lastOutputLock.RLock()
|
|
defer c.lastOutputLock.RUnlock()
|
|
|
|
const prefix = "TTL expired"
|
|
if c.lastOutput == "" {
|
|
return prefix
|
|
}
|
|
|
|
return fmt.Sprintf("%s (last output before timeout follows): %s", prefix, c.lastOutput)
|
|
}
|
|
|
|
// SetStatus is used to update the status of the check,
|
|
// and to renew the TTL. If expired, TTL is restarted.
|
|
// output is returned (might be truncated)
|
|
func (c *CheckTTL) SetStatus(status, output string) string {
|
|
c.Logger.Debug("Check status updated",
|
|
"check", c.CheckID.String(),
|
|
"status", status,
|
|
)
|
|
total := len(output)
|
|
if total > c.OutputMaxSize {
|
|
output = fmt.Sprintf("%s ... (captured %d of %d bytes)",
|
|
output[:c.OutputMaxSize], c.OutputMaxSize, total)
|
|
}
|
|
c.Notify.UpdateCheck(c.CheckID, status, output)
|
|
// Store the last output so we can retain it if the TTL expires.
|
|
c.lastOutputLock.Lock()
|
|
c.lastOutput = output
|
|
c.lastOutputLock.Unlock()
|
|
|
|
c.timer.Reset(c.TTL)
|
|
return output
|
|
}
|
|
|
|
// CheckHTTP is used to periodically make an HTTP request to
|
|
// determine the health of a given check.
|
|
// The check is passing if the response code is 2XX.
|
|
// The check is warning if the response code is 429.
|
|
// The check is critical if the response code is anything else
|
|
// or if the request returns an error
|
|
// Supports failures_before_critical and success_before_passing.
|
|
type CheckHTTP struct {
|
|
CheckID structs.CheckID
|
|
ServiceID structs.ServiceID
|
|
HTTP string
|
|
Header map[string][]string
|
|
Method string
|
|
Body string
|
|
Interval time.Duration
|
|
Timeout time.Duration
|
|
Logger hclog.Logger
|
|
TLSClientConfig *tls.Config
|
|
OutputMaxSize int
|
|
StatusHandler *StatusHandler
|
|
|
|
httpClient *http.Client
|
|
stop bool
|
|
stopCh chan struct{}
|
|
stopLock sync.Mutex
|
|
|
|
// Set if checks are exposed through Connect proxies
|
|
// If set, this is the target of check()
|
|
ProxyHTTP string
|
|
}
|
|
|
|
func (c *CheckHTTP) CheckType() structs.CheckType {
|
|
return structs.CheckType{
|
|
CheckID: c.CheckID.ID,
|
|
HTTP: c.HTTP,
|
|
Method: c.Method,
|
|
Body: c.Body,
|
|
Header: c.Header,
|
|
Interval: c.Interval,
|
|
ProxyHTTP: c.ProxyHTTP,
|
|
Timeout: c.Timeout,
|
|
OutputMaxSize: c.OutputMaxSize,
|
|
}
|
|
}
|
|
|
|
// Start is used to start an HTTP check.
|
|
// The check runs until stop is called
|
|
func (c *CheckHTTP) Start() {
|
|
c.stopLock.Lock()
|
|
defer c.stopLock.Unlock()
|
|
|
|
if c.httpClient == nil {
|
|
// Create the transport. We disable HTTP Keep-Alive's to prevent
|
|
// failing checks due to the keepalive interval.
|
|
trans := cleanhttp.DefaultTransport()
|
|
trans.DisableKeepAlives = true
|
|
|
|
// Take on the supplied TLS client config.
|
|
trans.TLSClientConfig = c.TLSClientConfig
|
|
|
|
// Create the HTTP client.
|
|
c.httpClient = &http.Client{
|
|
Timeout: 10 * time.Second,
|
|
Transport: trans,
|
|
}
|
|
if c.Timeout > 0 {
|
|
c.httpClient.Timeout = c.Timeout
|
|
}
|
|
|
|
if c.OutputMaxSize < 1 {
|
|
c.OutputMaxSize = DefaultBufSize
|
|
}
|
|
}
|
|
|
|
c.stop = false
|
|
c.stopCh = make(chan struct{})
|
|
go c.run()
|
|
}
|
|
|
|
// Stop is used to stop an HTTP check.
|
|
func (c *CheckHTTP) Stop() {
|
|
c.stopLock.Lock()
|
|
defer c.stopLock.Unlock()
|
|
if !c.stop {
|
|
c.stop = true
|
|
close(c.stopCh)
|
|
}
|
|
}
|
|
|
|
// run is invoked by a goroutine to run until Stop() is called
|
|
func (c *CheckHTTP) run() {
|
|
// Get the randomized initial pause time
|
|
initialPauseTime := lib.RandomStagger(c.Interval)
|
|
next := time.After(initialPauseTime)
|
|
for {
|
|
select {
|
|
case <-next:
|
|
c.check()
|
|
next = time.After(c.Interval)
|
|
case <-c.stopCh:
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// check is invoked periodically to perform the HTTP check
|
|
func (c *CheckHTTP) check() {
|
|
method := c.Method
|
|
if method == "" {
|
|
method = "GET"
|
|
}
|
|
|
|
target := c.HTTP
|
|
if c.ProxyHTTP != "" {
|
|
target = c.ProxyHTTP
|
|
}
|
|
|
|
bodyReader := strings.NewReader(c.Body)
|
|
req, err := http.NewRequest(method, target, bodyReader)
|
|
if err != nil {
|
|
c.StatusHandler.updateCheck(c.CheckID, api.HealthCritical, err.Error())
|
|
return
|
|
}
|
|
|
|
req.Header = http.Header(c.Header)
|
|
|
|
// this happens during testing but not in prod
|
|
if req.Header == nil {
|
|
req.Header = make(http.Header)
|
|
}
|
|
|
|
if host := req.Header.Get("Host"); host != "" {
|
|
req.Host = host
|
|
}
|
|
|
|
if req.Header.Get("User-Agent") == "" {
|
|
req.Header.Set("User-Agent", UserAgent)
|
|
}
|
|
if req.Header.Get("Accept") == "" {
|
|
req.Header.Set("Accept", "text/plain, text/*, */*")
|
|
}
|
|
|
|
resp, err := c.httpClient.Do(req)
|
|
if err != nil {
|
|
c.StatusHandler.updateCheck(c.CheckID, api.HealthCritical, err.Error())
|
|
return
|
|
}
|
|
defer resp.Body.Close()
|
|
|
|
// Read the response into a circular buffer to limit the size
|
|
output, _ := circbuf.NewBuffer(int64(c.OutputMaxSize))
|
|
if _, err := io.Copy(output, resp.Body); err != nil {
|
|
c.Logger.Warn("Check error while reading body",
|
|
"check", c.CheckID.String(),
|
|
"error", err,
|
|
)
|
|
}
|
|
|
|
// Format the response body
|
|
result := fmt.Sprintf("HTTP %s %s: %s Output: %s", method, target, resp.Status, output.String())
|
|
|
|
if resp.StatusCode >= 200 && resp.StatusCode <= 299 {
|
|
// PASSING (2xx)
|
|
c.StatusHandler.updateCheck(c.CheckID, api.HealthPassing, result)
|
|
} else if resp.StatusCode == 429 {
|
|
// WARNING
|
|
// 429 Too Many Requests (RFC 6585)
|
|
// The user has sent too many requests in a given amount of time.
|
|
c.StatusHandler.updateCheck(c.CheckID, api.HealthWarning, result)
|
|
} else {
|
|
// CRITICAL
|
|
c.StatusHandler.updateCheck(c.CheckID, api.HealthCritical, result)
|
|
}
|
|
}
|
|
|
|
// CheckTCP is used to periodically make an TCP/UDP connection to
|
|
// determine the health of a given check.
|
|
// The check is passing if the connection succeeds
|
|
// The check is critical if the connection returns an error
|
|
// Supports failures_before_critical and success_before_passing.
|
|
type CheckTCP struct {
|
|
CheckID structs.CheckID
|
|
ServiceID structs.ServiceID
|
|
TCP string
|
|
Interval time.Duration
|
|
Timeout time.Duration
|
|
Logger hclog.Logger
|
|
StatusHandler *StatusHandler
|
|
|
|
dialer *net.Dialer
|
|
stop bool
|
|
stopCh chan struct{}
|
|
stopLock sync.Mutex
|
|
}
|
|
|
|
// Start is used to start a TCP check.
|
|
// The check runs until stop is called
|
|
func (c *CheckTCP) Start() {
|
|
c.stopLock.Lock()
|
|
defer c.stopLock.Unlock()
|
|
|
|
if c.dialer == nil {
|
|
// Create the socket dialer
|
|
c.dialer = &net.Dialer{
|
|
Timeout: 10 * time.Second,
|
|
DualStack: true,
|
|
}
|
|
if c.Timeout > 0 {
|
|
c.dialer.Timeout = c.Timeout
|
|
}
|
|
}
|
|
|
|
c.stop = false
|
|
c.stopCh = make(chan struct{})
|
|
go c.run()
|
|
}
|
|
|
|
// Stop is used to stop a TCP check.
|
|
func (c *CheckTCP) Stop() {
|
|
c.stopLock.Lock()
|
|
defer c.stopLock.Unlock()
|
|
if !c.stop {
|
|
c.stop = true
|
|
close(c.stopCh)
|
|
}
|
|
}
|
|
|
|
// run is invoked by a goroutine to run until Stop() is called
|
|
func (c *CheckTCP) run() {
|
|
// Get the randomized initial pause time
|
|
initialPauseTime := lib.RandomStagger(c.Interval)
|
|
next := time.After(initialPauseTime)
|
|
for {
|
|
select {
|
|
case <-next:
|
|
c.check()
|
|
next = time.After(c.Interval)
|
|
case <-c.stopCh:
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
// check is invoked periodically to perform the TCP check
|
|
func (c *CheckTCP) check() {
|
|
conn, err := c.dialer.Dial(`tcp`, c.TCP)
|
|
if err != nil {
|
|
c.Logger.Warn("Check socket connection failed",
|
|
"check", c.CheckID.String(),
|
|
"error", err,
|
|
)
|
|
c.StatusHandler.updateCheck(c.CheckID, api.HealthCritical, err.Error())
|
|
return
|
|
}
|
|
conn.Close()
|
|
c.StatusHandler.updateCheck(c.CheckID, api.HealthPassing, fmt.Sprintf("TCP connect %s: Success", c.TCP))
|
|
}
|
|
|
|
// CheckDocker is used to periodically invoke a script to
|
|
// determine the health of an application running inside a
|
|
// Docker Container. We assume that the script is compatible
|
|
// with nagios plugins and expects the output in the same format.
|
|
// Supports failures_before_critical and success_before_passing.
|
|
type CheckDocker struct {
|
|
CheckID structs.CheckID
|
|
ServiceID structs.ServiceID
|
|
Script string
|
|
ScriptArgs []string
|
|
DockerContainerID string
|
|
Shell string
|
|
Interval time.Duration
|
|
Logger hclog.Logger
|
|
Client *DockerClient
|
|
StatusHandler *StatusHandler
|
|
|
|
stop chan struct{}
|
|
}
|
|
|
|
func (c *CheckDocker) Start() {
|
|
if c.stop != nil {
|
|
panic("Docker check already started")
|
|
}
|
|
|
|
if c.Logger == nil {
|
|
c.Logger = testutil.NewDiscardLogger()
|
|
}
|
|
|
|
if c.Shell == "" {
|
|
c.Shell = os.Getenv("SHELL")
|
|
if c.Shell == "" {
|
|
c.Shell = "/bin/sh"
|
|
}
|
|
}
|
|
c.stop = make(chan struct{})
|
|
go c.run()
|
|
}
|
|
|
|
func (c *CheckDocker) Stop() {
|
|
if c.stop == nil {
|
|
panic("Stop called before start")
|
|
}
|
|
close(c.stop)
|
|
}
|
|
|
|
func (c *CheckDocker) run() {
|
|
defer c.Client.Close()
|
|
firstWait := lib.RandomStagger(c.Interval)
|
|
next := time.After(firstWait)
|
|
for {
|
|
select {
|
|
case <-next:
|
|
c.check()
|
|
next = time.After(c.Interval)
|
|
case <-c.stop:
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
func (c *CheckDocker) check() {
|
|
var out string
|
|
status, b, err := c.doCheck()
|
|
if err != nil {
|
|
c.Logger.Debug("Check failed",
|
|
"check", c.CheckID.String(),
|
|
"error", err,
|
|
)
|
|
out = err.Error()
|
|
} else {
|
|
// out is already limited to CheckBufSize since we're getting a
|
|
// limited buffer. So we don't need to truncate it just report
|
|
// that it was truncated.
|
|
out = string(b.Bytes())
|
|
if int(b.TotalWritten()) > len(out) {
|
|
out = fmt.Sprintf("Captured %d of %d bytes\n...\n%s", len(out), b.TotalWritten(), out)
|
|
}
|
|
c.Logger.Trace("Check output",
|
|
"check", c.CheckID.String(),
|
|
"output", out,
|
|
)
|
|
}
|
|
c.StatusHandler.updateCheck(c.CheckID, status, out)
|
|
}
|
|
|
|
func (c *CheckDocker) doCheck() (string, *circbuf.Buffer, error) {
|
|
var cmd []string
|
|
if len(c.ScriptArgs) > 0 {
|
|
cmd = c.ScriptArgs
|
|
} else {
|
|
cmd = []string{c.Shell, "-c", c.Script}
|
|
}
|
|
|
|
execID, err := c.Client.CreateExec(c.DockerContainerID, cmd)
|
|
if err != nil {
|
|
return api.HealthCritical, nil, err
|
|
}
|
|
|
|
buf, err := c.Client.StartExec(c.DockerContainerID, execID)
|
|
if err != nil {
|
|
return api.HealthCritical, nil, err
|
|
}
|
|
|
|
exitCode, err := c.Client.InspectExec(c.DockerContainerID, execID)
|
|
if err != nil {
|
|
return api.HealthCritical, nil, err
|
|
}
|
|
|
|
switch exitCode {
|
|
case 0:
|
|
return api.HealthPassing, buf, nil
|
|
case 1:
|
|
c.Logger.Debug("Check failed",
|
|
"check", c.CheckID.String(),
|
|
"exit_code", exitCode,
|
|
)
|
|
return api.HealthWarning, buf, nil
|
|
default:
|
|
c.Logger.Debug("Check failed",
|
|
"check", c.CheckID.String(),
|
|
"exit_code", exitCode,
|
|
)
|
|
return api.HealthCritical, buf, nil
|
|
}
|
|
}
|
|
|
|
// CheckGRPC is used to periodically send request to a gRPC server
|
|
// application that implements gRPC health-checking protocol.
|
|
// The check is passing if returned status is SERVING.
|
|
// The check is critical if connection fails or returned status is
|
|
// not SERVING.
|
|
// Supports failures_before_critical and success_before_passing.
|
|
type CheckGRPC struct {
|
|
CheckID structs.CheckID
|
|
ServiceID structs.ServiceID
|
|
GRPC string
|
|
Interval time.Duration
|
|
Timeout time.Duration
|
|
TLSClientConfig *tls.Config
|
|
Logger hclog.Logger
|
|
StatusHandler *StatusHandler
|
|
|
|
probe *GrpcHealthProbe
|
|
stop bool
|
|
stopCh chan struct{}
|
|
stopLock sync.Mutex
|
|
|
|
// Set if checks are exposed through Connect proxies
|
|
// If set, this is the target of check()
|
|
ProxyGRPC string
|
|
}
|
|
|
|
func (c *CheckGRPC) CheckType() structs.CheckType {
|
|
return structs.CheckType{
|
|
CheckID: c.CheckID.ID,
|
|
GRPC: c.GRPC,
|
|
ProxyGRPC: c.ProxyGRPC,
|
|
Interval: c.Interval,
|
|
Timeout: c.Timeout,
|
|
}
|
|
}
|
|
|
|
func (c *CheckGRPC) Start() {
|
|
c.stopLock.Lock()
|
|
defer c.stopLock.Unlock()
|
|
timeout := 10 * time.Second
|
|
if c.Timeout > 0 {
|
|
timeout = c.Timeout
|
|
}
|
|
c.probe = NewGrpcHealthProbe(c.GRPC, timeout, c.TLSClientConfig)
|
|
c.stop = false
|
|
c.stopCh = make(chan struct{})
|
|
go c.run()
|
|
}
|
|
|
|
func (c *CheckGRPC) run() {
|
|
// Get the randomized initial pause time
|
|
initialPauseTime := lib.RandomStagger(c.Interval)
|
|
next := time.After(initialPauseTime)
|
|
for {
|
|
select {
|
|
case <-next:
|
|
c.check()
|
|
next = time.After(c.Interval)
|
|
case <-c.stopCh:
|
|
return
|
|
}
|
|
}
|
|
}
|
|
|
|
func (c *CheckGRPC) check() {
|
|
target := c.GRPC
|
|
if c.ProxyGRPC != "" {
|
|
target = c.ProxyGRPC
|
|
}
|
|
|
|
err := c.probe.Check(target)
|
|
if err != nil {
|
|
c.StatusHandler.updateCheck(c.CheckID, api.HealthCritical, err.Error())
|
|
} else {
|
|
c.StatusHandler.updateCheck(c.CheckID, api.HealthPassing, fmt.Sprintf("gRPC check %s: success", target))
|
|
}
|
|
}
|
|
|
|
func (c *CheckGRPC) Stop() {
|
|
c.stopLock.Lock()
|
|
defer c.stopLock.Unlock()
|
|
if !c.stop {
|
|
c.stop = true
|
|
close(c.stopCh)
|
|
}
|
|
}
|
|
|
|
// StatusHandler keep tracks of successive error/success counts and ensures
|
|
// that status can be set to critical/passing only once the successive number of event
|
|
// reaches the given threshold.
|
|
type StatusHandler struct {
|
|
inner CheckNotifier
|
|
logger hclog.Logger
|
|
successBeforePassing int
|
|
successCounter int
|
|
failuresBeforeCritical int
|
|
failuresCounter int
|
|
}
|
|
|
|
// NewStatusHandler set counters values to threshold in order to immediatly update status after first check.
|
|
func NewStatusHandler(inner CheckNotifier, logger hclog.Logger, successBeforePassing, failuresBeforeCritical int) *StatusHandler {
|
|
return &StatusHandler{
|
|
logger: logger,
|
|
inner: inner,
|
|
successBeforePassing: successBeforePassing,
|
|
successCounter: successBeforePassing,
|
|
failuresBeforeCritical: failuresBeforeCritical,
|
|
failuresCounter: failuresBeforeCritical,
|
|
}
|
|
}
|
|
|
|
func (s *StatusHandler) updateCheck(checkID structs.CheckID, status, output string) {
|
|
|
|
if status == api.HealthPassing || status == api.HealthWarning {
|
|
s.successCounter++
|
|
s.failuresCounter = 0
|
|
if s.successCounter >= s.successBeforePassing {
|
|
s.logger.Debug("Check status updated",
|
|
"check", checkID.String(),
|
|
"status", status,
|
|
)
|
|
s.inner.UpdateCheck(checkID, status, output)
|
|
return
|
|
}
|
|
s.logger.Warn("Check passed but has not reached success threshold",
|
|
"check", checkID.String(),
|
|
"status", status,
|
|
"success_count", s.successCounter,
|
|
"success_threshold", s.successBeforePassing,
|
|
)
|
|
} else {
|
|
s.failuresCounter++
|
|
s.successCounter = 0
|
|
if s.failuresCounter >= s.failuresBeforeCritical {
|
|
s.logger.Warn("Check is now critical", "check", checkID.String())
|
|
s.inner.UpdateCheck(checkID, status, output)
|
|
return
|
|
}
|
|
s.logger.Warn("Check failed but has not reached failure threshold",
|
|
"check", checkID.String(),
|
|
"status", status,
|
|
"failure_count", s.failuresCounter,
|
|
"failure_threshold", s.failuresBeforeCritical,
|
|
)
|
|
}
|
|
}
|