Feature: Health checks windows service

.changelog/13388.txt

@@ -0,0 +1 @@
+agent: windows service health check

agent/agent.go

@@ -264,6 +264,9 @@ type Agent struct {
 	// checkAliases maps the check ID to an associated Alias checks
 	checkAliases map[structs.CheckID]*checks.CheckAlias
 
+	// checkOSServices maps the check ID to an associated OS Service check
+	checkOSServices map[structs.CheckID]*checks.CheckOSService
+
 	// exposedPorts tracks listener ports for checks exposed through a proxy
 	exposedPorts map[string]int
 
@@ -273,6 +276,9 @@ type Agent struct {
 	// dockerClient is the client for performing docker health checks.
 	dockerClient *checks.DockerClient
 
+	// osServiceClient is the client for performing OS service checks.
+	osServiceClient *checks.OSServiceClient
+
 	// eventCh is used to receive user events
 	eventCh chan serf.UserEvent
 
@@ -409,6 +415,7 @@ func New(bd BaseDeps) (*Agent, error) {
 		checkGRPCs:      make(map[structs.CheckID]*checks.CheckGRPC),
 		checkDockers:    make(map[structs.CheckID]*checks.CheckDocker),
 		checkAliases:    make(map[structs.CheckID]*checks.CheckAlias),
+		checkOSServices: make(map[structs.CheckID]*checks.CheckOSService),
 		eventCh:         make(chan serf.UserEvent, 1024),
 		eventBuf:        make([]*UserEvent, 256),
 		joinLANNotifier: &systemd.Notifier{},
@@ -2894,12 +2901,45 @@ func (a *Agent) addCheck(check *structs.HealthCheck, chkType *structs.CheckType,
 				Client:            a.dockerClient,
 				StatusHandler:     statusHandler,
 			}
-			if prev := a.checkDockers[cid]; prev != nil {
-				prev.Stop()
-			}
 			dockerCheck.Start()
 			a.checkDockers[cid] = dockerCheck
 
+		case chkType.IsOSService():
+			if existing, ok := a.checkOSServices[cid]; ok {
+				existing.Stop()
+				delete(a.checkOSServices, cid)
+			}
+			if chkType.Interval < checks.MinInterval {
+				a.logger.Warn("check has interval below minimum",
+					"check", cid.String(),
+					"minimum_interval", checks.MinInterval,
+				)
+				chkType.Interval = checks.MinInterval
+			}
+
+			if a.osServiceClient == nil {
+				ossp, err := checks.NewOSServiceClient()
+				if err != nil {
+					a.logger.Error("error creating OS Service client", "error", err)
+					return err
+				}
+				a.logger.Debug("created OS Service client")
+				a.osServiceClient = ossp
+			}
+
+			osServiceCheck := &checks.CheckOSService{
+				CheckID:       cid,
+				ServiceID:     sid,
+				OSService:     chkType.OSService,
+				Timeout:       chkType.Timeout,
+				Interval:      chkType.Interval,
+				Logger:        a.logger,
+				Client:        a.osServiceClient,
+				StatusHandler: statusHandler,
+			}
+			osServiceCheck.Start()
+			a.checkOSServices[cid] = osServiceCheck
+
 		case chkType.IsMonitor():
 			if existing, ok := a.checkMonitors[cid]; ok {
 				existing.Stop()

agent/checks/check.go

@@ -4,6 +4,7 @@ import (
 	"bufio"
 	"context"
 	"crypto/tls"
+	"errors"
 	"fmt"
 	"io"
 	"io/ioutil"
@@ -1047,6 +1048,125 @@ func (c *CheckGRPC) Stop() {
 	}
 }
 
+type CheckOSService struct {
+	CheckID       structs.CheckID
+	ServiceID     structs.ServiceID
+	OSService     string
+	Interval      time.Duration
+	Timeout       time.Duration
+	Logger        hclog.Logger
+	StatusHandler *StatusHandler
+	Client        *OSServiceClient
+
+	stop     bool
+	stopCh   chan struct{}
+	stopLock sync.Mutex
+	stopWg   sync.WaitGroup
+}
+
+func (c *CheckOSService) CheckType() structs.CheckType {
+	return structs.CheckType{
+		CheckID:   c.CheckID.ID,
+		OSService: c.OSService,
+		Interval:  c.Interval,
+		Timeout:   c.Timeout,
+	}
+}
+
+func (c *CheckOSService) Start() {
+	c.stopLock.Lock()
+	defer c.stopLock.Unlock()
+	c.stop = false
+	c.stopCh = make(chan struct{})
+	c.stopWg.Add(1)
+	go c.run()
+}
+
+func (c *CheckOSService) Stop() {
+	c.stopLock.Lock()
+	defer c.stopLock.Unlock()
+	if !c.stop {
+		c.stop = true
+		close(c.stopCh)
+	}
+
+	// Wait for the c.run() goroutine to complete before returning.
+	c.stopWg.Wait()
+}
+
+func (c *CheckOSService) run() {
+	defer c.stopWg.Done()
+	// 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 *CheckOSService) doCheck() (string, error) {
+	err := c.Client.Check(c.OSService)
+	if err == nil {
+		return api.HealthPassing, nil
+	}
+	if errors.Is(err, ErrOSServiceStatusCritical) {
+		return api.HealthCritical, err
+	}
+
+	return api.HealthWarning, err
+}
+
+func (c *CheckOSService) check() {
+	var out string
+	var status string
+	var err error
+
+	waitCh := make(chan error, 1)
+	go func() {
+		status, err = c.doCheck()
+		waitCh <- err
+	}()
+
+	timeout := 30 * time.Second
+	if c.Timeout > 0 {
+		timeout = c.Timeout
+	}
+	select {
+	case <-time.After(timeout):
+		msg := fmt.Sprintf("Timed out (%s) running check", timeout.String())
+		c.Logger.Warn("Timed out running check",
+			"check", c.CheckID.String(),
+			"timeout", timeout.String(),
+		)
+
+		c.StatusHandler.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
+	}
+
+	out = fmt.Sprintf("Service \"%s\" is healthy", c.OSService)
+	if err != nil {
+		c.Logger.Debug("Check failed",
+			"check", c.CheckID.String(),
+			"error", err,
+		)
+		out = err.Error()
+	}
+	c.StatusHandler.updateCheck(c.CheckID, status, out)
+}
+
 // 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.