Merge pull request #4251 from hashicorp/f-grpc-checks

Support Consul gRPC Health Checks

CHANGELOG.md

@@ -2,8 +2,9 @@
 
 IMPROVEMENTS:
  * cli: Add node drain details to node status [[GH-4247](https://github.com/hashicorp/nomad/issues/4247)]
- * command: add -short option to init command that emits a minimal
+ * command: Add -short option to init command that emits a minimal
    jobspec [[GH-4239](https://github.com/hashicorp/nomad/issues/4239)]
+ * discovery: Support Consul gRPC health checks. [[GH-4251](https://github.com/hashicorp/nomad/issues/4251)]
 
 ## 0.8.3 (April 27, 2018)
 

api/tasks.go

@@ -286,6 +286,8 @@ type ServiceCheck struct {
 	Header        map[string][]string
 	Method        string
 	CheckRestart  *CheckRestart `mapstructure:"check_restart"`
+	GRPCService   string        `mapstructure:"grpc_service"`
+	GRPCUseTLS    bool          `mapstructure:"grpc_use_tls"`
 }
 
 // The Service model represents a Consul service definition

client/task_runner.go

@@ -1516,6 +1516,7 @@ func interpolateServices(taskEnv *env.TaskEnv, task *structs.Task) *structs.Task
 			check.PortLabel = taskEnv.ReplaceEnv(check.PortLabel)
 			check.InitialStatus = taskEnv.ReplaceEnv(check.InitialStatus)
 			check.Method = taskEnv.ReplaceEnv(check.Method)
+			check.GRPCService = taskEnv.ReplaceEnv(check.GRPCService)
 			if len(check.Header) > 0 {
 				header := make(map[string][]string, len(check.Header))
 				for k, vs := range check.Header {

command/agent/consul/client.go

@@ -1094,12 +1094,22 @@ func createCheckReg(serviceID, checkID string, check *structs.ServiceCheck, host
 		chkReg.HTTP = url.String()
 		chkReg.Method = check.Method
 		chkReg.Header = check.Header
+
 	case structs.ServiceCheckTCP:
 		chkReg.TCP = net.JoinHostPort(host, strconv.Itoa(port))
+
 	case structs.ServiceCheckScript:
 		chkReg.TTL = (check.Interval + ttlCheckBuffer).String()
 		// As of Consul 1.0.0 setting TTL and Interval is a 400
 		chkReg.Interval = ""
+
+	case structs.ServiceCheckGRPC:
+		chkReg.GRPC = fmt.Sprintf("%s/%s", net.JoinHostPort(host, strconv.Itoa(port)), check.GRPCService)
+		chkReg.GRPCUseTLS = check.GRPCUseTLS
+		if check.TLSSkipVerify {
+			chkReg.TLSSkipVerify = true
+		}
+
 	default:
 		return nil, fmt.Errorf("check type %+q not valid", check.Type)
 	}

command/agent/consul/unit_test.go

@@ -1391,9 +1391,10 @@ func TestIsNomadService(t *testing.T) {
 	}
 }
 
-// TestCreateCheckReg asserts Nomad ServiceCheck structs are properly converted
-// to Consul API AgentCheckRegistrations.
-func TestCreateCheckReg(t *testing.T) {
+// TestCreateCheckReg_HTTP asserts Nomad ServiceCheck structs are properly
+// converted to Consul API AgentCheckRegistrations for HTTP checks.
+func TestCreateCheckReg_HTTP(t *testing.T) {
+	t.Parallel()
 	check := &structs.ServiceCheck{
 		Name:      "name",
 		Type:      "http",
@@ -1435,6 +1436,42 @@ func TestCreateCheckReg(t *testing.T) {
 	}
 }
 
+// TestCreateCheckReg_GRPC asserts Nomad ServiceCheck structs are properly
+// converted to Consul API AgentCheckRegistrations for GRPC checks.
+func TestCreateCheckReg_GRPC(t *testing.T) {
+	t.Parallel()
+	check := &structs.ServiceCheck{
+		Name:          "name",
+		Type:          "grpc",
+		PortLabel:     "label",
+		GRPCService:   "foo.Bar",
+		GRPCUseTLS:    true,
+		TLSSkipVerify: true,
+		Timeout:       time.Second,
+		Interval:      time.Minute,
+	}
+
+	serviceID := "testService"
+	checkID := check.Hash(serviceID)
+
+	expected := &api.AgentCheckRegistration{
+		ID:        checkID,
+		Name:      "name",
+		ServiceID: serviceID,
+		AgentServiceCheck: api.AgentServiceCheck{
+			Timeout:       "1s",
+			Interval:      "1m0s",
+			GRPC:          "localhost:8080/foo.Bar",
+			GRPCUseTLS:    true,
+			TLSSkipVerify: true,
+		},
+	}
+
+	actual, err := createCheckReg(serviceID, checkID, check, "localhost", 8080)
+	require.NoError(t, err)
+	require.Equal(t, expected, actual)
+}
+
 // TestGetAddress asserts Nomad uses the correct ip and port for services and
 // checks depending on port labels, driver networks, and address mode.
 func TestGetAddress(t *testing.T) {

command/agent/job_endpoint.go

@@ -764,6 +764,8 @@ func ApiTaskToStructsTask(apiTask *api.Task, structsTask *structs.Task) {
 						TLSSkipVerify: check.TLSSkipVerify,
 						Header:        check.Header,
 						Method:        check.Method,
+						GRPCService:   check.GRPCService,
+						GRPCUseTLS:    check.GRPCUseTLS,
 					}
 					if check.CheckRestart != nil {
 						structsTask.Services[i].Checks[j].CheckRestart = &structs.CheckRestart{

command/agent/job_endpoint_test.go

@@ -1272,6 +1272,8 @@ func TestJobs_ApiJobToStructsJob(t *testing.T) {
 										Protocol:      "http",
 										PortLabel:     "foo",
 										AddressMode:   "driver",
+										GRPCService:   "foo.Bar",
+										GRPCUseTLS:    true,
 										Interval:      4 * time.Second,
 										Timeout:       2 * time.Second,
 										InitialStatus: "ok",
@@ -1493,6 +1495,8 @@ func TestJobs_ApiJobToStructsJob(t *testing.T) {
 										Interval:      4 * time.Second,
 										Timeout:       2 * time.Second,
 										InitialStatus: "ok",
+										GRPCService:   "foo.Bar",
+										GRPCUseTLS:    true,
 										CheckRestart: &structs.CheckRestart{
 											Limit:          3,
 											Grace:          11 * time.Second,

jobspec/parse.go

@@ -1061,6 +1061,8 @@ func parseChecks(service *api.Service, checkObjs *ast.ObjectList) error {
 			"method",
 			"check_restart",
 			"address_mode",
+			"grpc_service",
+			"grpc_use_tls",
 		}
 		if err := helper.CheckHCLKeys(co.Val, valid); err != nil {
 			return multierror.Prefix(err, "check ->")

jobspec/parse_test.go

@@ -135,11 +135,13 @@ func TestParse(t *testing.T) {
 										PortLabel: "http",
 										Checks: []api.ServiceCheck{
 											{
-												Name:      "check-name",
-												Type:      "tcp",
-												PortLabel: "admin",
-												Interval:  10 * time.Second,
-												Timeout:   2 * time.Second,
+												Name:        "check-name",
+												Type:        "tcp",
+												PortLabel:   "admin",
+												Interval:    10 * time.Second,
+												Timeout:     2 * time.Second,
+												GRPCService: "foo.Bar",
+												GRPCUseTLS:  true,
 												CheckRestart: &api.CheckRestart{
 													Limit:          3,
 													Grace:          helper.TimeToPtr(10 * time.Second),

jobspec/test-fixtures/basic.hcl

@@ -102,11 +102,13 @@ job "binstore-storagelocker" {
         port = "http"
 
         check {
-          name     = "check-name"
-          type     = "tcp"
-          interval = "10s"
-          timeout  = "2s"
-          port     = "admin"
+          name         = "check-name"
+          type         = "tcp"
+          interval     = "10s"
+          timeout      = "2s"
+          port         = "admin"
+          grpc_service = "foo.Bar"
+          grpc_use_tls = true
 
           check_restart {
             limit = 3

nomad/structs/diff_test.go

@@ -3557,6 +3557,12 @@ func TestTaskDiff(t *testing.T) {
 										Old:  "",
 										New:  "foo",
 									},
+									{
+										Type: DiffTypeAdded,
+										Name: "GRPCUseTLS",
+										Old:  "",
+										New:  "false",
+									},
 									{
 										Type: DiffTypeAdded,
 										Name: "Interval",
@@ -3611,6 +3617,12 @@ func TestTaskDiff(t *testing.T) {
 										Old:  "foo",
 										New:  "",
 									},
+									{
+										Type: DiffTypeDeleted,
+										Name: "GRPCUseTLS",
+										Old:  "false",
+										New:  "",
+									},
 									{
 										Type: DiffTypeDeleted,
 										Name: "Interval",
@@ -3767,6 +3779,18 @@ func TestTaskDiff(t *testing.T) {
 										Old:  "foo",
 										New:  "foo",
 									},
+									{
+										Type: DiffTypeNone,
+										Name: "GRPCService",
+										Old:  "",
+										New:  "",
+									},
+									{
+										Type: DiffTypeNone,
+										Name: "GRPCUseTLS",
+										Old:  "false",
+										New:  "false",
+									},
 									{
 										Type: DiffTypeEdited,
 										Name: "InitialStatus",

nomad/structs/structs.go

@@ -3515,6 +3515,7 @@ const (
 	ServiceCheckHTTP   = "http"
 	ServiceCheckTCP    = "tcp"
 	ServiceCheckScript = "script"
+	ServiceCheckGRPC   = "grpc"
 
 	// minCheckInterval is the minimum check interval permitted.  Consul
 	// currently has its MinInterval set to 1s.  Mirror that here for
@@ -3544,6 +3545,8 @@ type ServiceCheck struct {
 	Method        string              // HTTP Method to use (GET by default)
 	Header        map[string][]string // HTTP Headers for Consul to set when making HTTP checks
 	CheckRestart  *CheckRestart       // If and when a task should be restarted based on checks
+	GRPCService   string              // Service for GRPC checks
+	GRPCUseTLS    bool                // Whether or not to use TLS for GRPC checks
 }
 
 func (sc *ServiceCheck) Copy() *ServiceCheck {
@@ -3584,6 +3587,7 @@ func (sc *ServiceCheck) Canonicalize(serviceName string) {
 func (sc *ServiceCheck) validate() error {
 	// Validate Type
 	switch strings.ToLower(sc.Type) {
+	case ServiceCheckGRPC:
 	case ServiceCheckTCP:
 	case ServiceCheckHTTP:
 		if sc.Path == "" {
@@ -3601,6 +3605,7 @@ func (sc *ServiceCheck) validate() error {
 		if sc.Command == "" {
 			return fmt.Errorf("script type must have a valid script path")
 		}
+
 	default:
 		return fmt.Errorf(`invalid type (%+q), must be one of "http", "tcp", or "script" type`, sc.Type)
 	}
@@ -3645,7 +3650,7 @@ func (sc *ServiceCheck) validate() error {
 // RequiresPort returns whether the service check requires the task has a port.
 func (sc *ServiceCheck) RequiresPort() bool {
 	switch sc.Type {
-	case ServiceCheckHTTP, ServiceCheckTCP:
+	case ServiceCheckGRPC, ServiceCheckHTTP, ServiceCheckTCP:
 		return true
 	default:
 		return false
@@ -3696,6 +3701,14 @@ func (sc *ServiceCheck) Hash(serviceID string) string {
 		io.WriteString(h, sc.AddressMode)
 	}
 
+	// Only include GRPC if set to maintain ID stability with Nomad <0.8.4
+	if sc.GRPCService != "" {
+		io.WriteString(h, sc.GRPCService)
+	}
+	if sc.GRPCUseTLS {
+		io.WriteString(h, "true")
+	}
+
 	return fmt.Sprintf("%x", h.Sum(nil))
 }
 

nomad/structs/structs_test.go

@@ -1267,7 +1267,34 @@ func TestTask_Validate_Service_Check_AddressMode(t *testing.T) {
 	}
 }
 
+func TestTask_Validate_Service_Check_GRPC(t *testing.T) {
+	t.Parallel()
+	// Bad (no port)
+	invalidGRPC := &ServiceCheck{
+		Type:     ServiceCheckGRPC,
+		Interval: time.Second,
+		Timeout:  time.Second,
+	}
+	service := &Service{
+		Name:   "test",
+		Checks: []*ServiceCheck{invalidGRPC},
+	}
+
+	assert.Error(t, service.Validate())
+
+	// Good
+	service.Checks[0] = &ServiceCheck{
+		Type:      ServiceCheckGRPC,
+		Interval:  time.Second,
+		Timeout:   time.Second,
+		PortLabel: "some-port-label",
+	}
+
+	assert.NoError(t, service.Validate())
+}
+
 func TestTask_Validate_Service_Check_CheckRestart(t *testing.T) {
+	t.Parallel()
 	invalidCheckRestart := &CheckRestart{
 		Limit: -1,
 		Grace: -1,