MINOR: consul: add support for health check policy

Author: rmaticevic

discovery/consul_service_discovery_instance.go

@@ -170,6 +170,9 @@ func (c *consulInstance) updateServices() error {
 func (c *consulInstance) convertToServers(nodes []*serviceEntry) []configuration.ServiceServer {
 	servers := make([]configuration.ServiceServer, 0)
 	for _, node := range nodes {
+		if !c.validateHealthChecks(node) {
+			continue
+		}
 		if node.Service.Address != "" {
 			servers = append(servers, configuration.ServiceServer{
 				Address: node.Service.Address,
@@ -185,6 +188,61 @@ func (c *consulInstance) convertToServers(nodes []*serviceEntry) []configuration
 	return servers
 }
 
+func (c *consulInstance) validateHealthChecks(node *serviceEntry) bool {
+	switch *c.params.HealthCheckPolicy {
+	case models.ConsulHealthCheckPolicyAny:
+		return c.validateHealthChecksAny(node)
+	case models.ConsulHealthCheckPolicyAll:
+		return c.validateHealthChecksAll(node)
+	case models.ConsulHealthCheckPolicyMin:
+		return c.validateHealthChecksMin(node)
+	case models.ConsulHealthCheckPolicyNone:
+		return true
+	default:
+		return true
+	}
+}
+
+func (c *consulInstance) validateHealthChecksAny(node *serviceEntry) bool {
+	if node.Checks == nil || len(node.Checks) == 0 {
+		return false
+	}
+
+	for _, check := range node.Checks {
+		if check.Status == "passing" {
+			return true
+		}
+	}
+	return false
+}
+
+func (c *consulInstance) validateHealthChecksAll(node *serviceEntry) bool {
+	if node.Checks == nil || len(node.Checks) == 0 {
+		return false
+	}
+
+	for _, check := range node.Checks {
+		if check.Status != "passing" {
+			return false
+		}
+	}
+	return true
+}
+
+func (c *consulInstance) validateHealthChecksMin(node *serviceEntry) bool {
+	if node.Checks == nil || len(node.Checks) == 0 {
+		return false
+	}
+
+	passing := 0
+	for _, check := range node.Checks {
+		if check.Status == "passing" {
+			passing++
+		}
+	}
+	return passing >= int(c.params.HealthCheckPolicyMin)
+}
+
 func (c *consulInstance) hasServiceChanged(service string, index uint64) bool {
 	prevIndex, ok := c.prevIndexes[service]
 	if !ok {
@@ -313,6 +371,9 @@ type serviceEntry struct {
 		Address string
 		Port    int
 	}
+	Checks []*struct {
+		Status string
+	}
 }
 
 type queryParams struct {

handlers/consul.go

@@ -64,6 +64,13 @@ func (c *CreateConsulHandlerImpl) Handle(params service_discovery.CreateConsulPa
 		return service_discovery.NewCreateConsulDefault(int(*e.Code)).WithPayload(e)
 	}
 	setFilters(params.Data)
+	if *params.Data.HealthCheckPolicy == models.ConsulHealthCheckPolicyMin && params.Data.HealthCheckPolicyMin <= 0 {
+		e := &models.Error{
+			Message: misc.StringP("health_check_policy_min is required for 'min' health_check_policy"),
+			Code:    misc.Int64P(int(misc.ErrHTTPBadRequest)),
+		}
+		return service_discovery.NewCreateConsulDefault(int(*e.Code)).WithPayload(e)
+	}
 	err := c.Discovery.AddNode("consul", *params.Data.ID, params.Data)
 	if err != nil {
 		e := misc.HandleError(err)
@@ -134,6 +141,13 @@ func (c *ReplaceConsulHandlerImpl) Handle(params service_discovery.ReplaceConsul
 		return service_discovery.NewReplaceConsulDefault(int(*e.Code)).WithPayload(e)
 	}
 	setFilters(params.Data)
+	if *params.Data.HealthCheckPolicy == models.ConsulHealthCheckPolicyMin && params.Data.HealthCheckPolicyMin <= 0 {
+		e := &models.Error{
+			Message: misc.StringP("health_check_policy_min is required for 'min' health_check_policy"),
+			Code:    misc.Int64P(int(misc.ErrHTTPBadRequest)),
+		}
+		return service_discovery.NewCreateConsulDefault(int(*e.Code)).WithPayload(e)
+	}
 	err := c.Discovery.UpdateNode("consul", *params.Data.ID, params.Data)
 	if err != nil {
 		e := misc.HandleError(err)