Merge pull request #1839 from hashicorp/f-signal-constraints

Signal creates an auto-constraints

Author: dadgar

client/driver/docker.go

@@ -267,6 +267,12 @@ func (d *DockerDriver) Validate(config map[string]interface{}) error {
 	return nil
 }
 
+func (d *DockerDriver) Abilities() DriverAbilities {
+	return DriverAbilities{
+		SendSignals: true,
+	}
+}
+
 // dockerClients creates two *docker.Client, one for long running operations and
 // the other for shorter operations. In test / dev mode we can use ENV vars to
 // connect to the docker daemon. In production mode we will read docker.endpoint

client/driver/driver.go

@@ -59,6 +59,15 @@ type Driver interface {
 
 	// Drivers must validate their configuration
 	Validate(map[string]interface{}) error
+
+	// Abilities returns the abilities of the driver
+	Abilities() DriverAbilities
+}
+
+// DriverAbilities marks the abilities the driver has.
+type DriverAbilities struct {
+	// SendSignals marks the driver as being able to send signals
+	SendSignals bool
 }
 
 // DriverContext is a means to inject dependencies such as loggers, configs, and

client/driver/exec.go

@@ -82,6 +82,12 @@ func (d *ExecDriver) Validate(config map[string]interface{}) error {
 	return nil
 }
 
+func (d *ExecDriver) Abilities() DriverAbilities {
+	return DriverAbilities{
+		SendSignals: true,
+	}
+}
+
 func (d *ExecDriver) Periodic() (bool, time.Duration) {
 	return true, 15 * time.Second
 }

client/driver/java.go

@@ -94,6 +94,12 @@ func (d *JavaDriver) Validate(config map[string]interface{}) error {
 	return nil
 }
 
+func (d *JavaDriver) Abilities() DriverAbilities {
+	return DriverAbilities{
+		SendSignals: true,
+	}
+}
+
 func (d *JavaDriver) Fingerprint(cfg *config.Config, node *structs.Node) (bool, error) {
 	// Get the current status so that we can log any debug messages only if the
 	// state changes

client/driver/mock_driver.go

@@ -62,6 +62,12 @@ func NewMockDriver(ctx *DriverContext) Driver {
 	return &MockDriver{DriverContext: *ctx}
 }
 
+func (d *MockDriver) Abilities() DriverAbilities {
+	return DriverAbilities{
+		SendSignals: false,
+	}
+}
+
 // Start starts the mock driver
 func (m *MockDriver) Start(ctx *ExecContext, task *structs.Task) (DriverHandle, error) {
 	var driverConfig MockDriverConfig

client/driver/qemu.go

@@ -97,6 +97,12 @@ func (d *QemuDriver) Validate(config map[string]interface{}) error {
 	return nil
 }
 
+func (d *QemuDriver) Abilities() DriverAbilities {
+	return DriverAbilities{
+		SendSignals: false,
+	}
+}
+
 func (d *QemuDriver) Fingerprint(cfg *config.Config, node *structs.Node) (bool, error) {
 	// Get the current status so that we can log any debug messages only if the
 	// state changes

client/driver/raw_exec.go

@@ -81,6 +81,12 @@ func (d *RawExecDriver) Validate(config map[string]interface{}) error {
 	return nil
 }
 
+func (d *RawExecDriver) Abilities() DriverAbilities {
+	return DriverAbilities{
+		SendSignals: true,
+	}
+}
+
 func (d *RawExecDriver) Fingerprint(cfg *config.Config, node *structs.Node) (bool, error) {
 	// Get the current status so that we can log any debug messages only if the
 	// state changes

client/driver/rkt.go

@@ -135,6 +135,12 @@ func (d *RktDriver) Validate(config map[string]interface{}) error {
 	return nil
 }
 
+func (d *RktDriver) Abilities() DriverAbilities {
+	return DriverAbilities{
+		SendSignals: false,
+	}
+}
+
 func (d *RktDriver) Fingerprint(cfg *config.Config, node *structs.Node) (bool, error) {
 	// Get the current status so that we can log any debug messages only if the
 	// state changes

client/fingerprint/fingerprint.go

@@ -25,6 +25,7 @@ func init() {
 	builtinFingerprintMap["memory"] = NewMemoryFingerprint
 	builtinFingerprintMap["network"] = NewNetworkFingerprint
 	builtinFingerprintMap["nomad"] = NewNomadFingerprint
+	builtinFingerprintMap["signal"] = NewSignalFingerprint
 	builtinFingerprintMap["storage"] = NewStorageFingerprint
 	builtinFingerprintMap["vault"] = NewVaultFingerprint
 

client/fingerprint/signal.go

@@ -0,0 +1,33 @@
+package fingerprint
+
+import (
+	"log"
+	"strings"
+
+	"github.com/hashicorp/consul-template/signals"
+	"github.com/hashicorp/nomad/client/config"
+	"github.com/hashicorp/nomad/nomad/structs"
+)
+
+// SignalFingerprint is used to fingerprint the available signals
+type SignalFingerprint struct {
+	StaticFingerprinter
+	logger *log.Logger
+}
+
+// NewSignalFingerprint is used to create a Signal fingerprint
+func NewSignalFingerprint(logger *log.Logger) Fingerprint {
+	f := &SignalFingerprint{logger: logger}
+	return f
+}
+
+func (f *SignalFingerprint) Fingerprint(cfg *config.Config, node *structs.Node) (bool, error) {
+	// Build the list of available signals
+	sigs := make([]string, 0, len(signals.SignalLookup))
+	for signal := range signals.SignalLookup {
+		sigs = append(sigs, signal)
+	}
+
+	node.Attributes["os.signals"] = strings.Join(sigs, ",")
+	return true, nil
+}

client/fingerprint/signal_test.go

@@ -0,0 +1,17 @@
+package fingerprint
+
+import (
+	"testing"
+
+	"github.com/hashicorp/nomad/nomad/structs"
+)
+
+func TestSignalFingerprint(t *testing.T) {
+	fp := NewSignalFingerprint(testLogger())
+	node := &structs.Node{
+		Attributes: make(map[string]string),
+	}
+
+	assertFingerprintOK(t, fp, node)
+	assertNodeAttributeContains(t, node, "os.signals")
+}

jobspec/parse.go

@@ -407,6 +407,7 @@ func parseConstraints(result *[]*structs.Constraint, list *ast.ObjectList) error
 			"version",
 			"regexp",
 			"distinct_hosts",
+			"set_contains",
 		}
 		if err := checkHCLKeys(o.Val, valid); err != nil {
 			return err
@@ -435,6 +436,13 @@ func parseConstraints(result *[]*structs.Constraint, list *ast.ObjectList) error
 			m["RTarget"] = constraint
 		}
 
+		// If "set_contains" is provided, set the operand
+		// to "set_contains" and the value to the "RTarget"
+		if constraint, ok := m[structs.ConstraintSetContains]; ok {
+			m["Operand"] = structs.ConstraintSetContains
+			m["RTarget"] = constraint
+		}
+
 		if value, ok := m[structs.ConstraintDistinctHosts]; ok {
 			enabled, err := parseBool(value)
 			if err != nil {

jobspec/parse_test.go

@@ -282,6 +282,25 @@ func TestParse(t *testing.T) {
 			false,
 		},
 
+		{
+			"set-contains-constraint.hcl",
+			&structs.Job{
+				ID:       "foo",
+				Name:     "foo",
+				Priority: 50,
+				Region:   "global",
+				Type:     "service",
+				Constraints: []*structs.Constraint{
+					&structs.Constraint{
+						LTarget: "$meta.data",
+						RTarget: "foo,bar,baz",
+						Operand: structs.ConstraintSetContains,
+					},
+				},
+			},
+			false,
+		},
+
 		{
 			"distinctHosts-constraint.hcl",
 			&structs.Job{

jobspec/test-fixtures/set-contains-constraint.hcl

@@ -0,0 +1,6 @@
+job "foo" {
+    constraint {
+        attribute = "$meta.data"
+        set_contains = "foo,bar,baz"
+    }
+}

nomad/job_endpoint.go

@@ -52,6 +52,9 @@ func (j *Job) Register(args *structs.JobRegisterRequest, reply *structs.JobRegis
 	// Initialize the job fields (sets defaults and any necessary init work).
 	args.Job.Canonicalize()
 
+	// Add implicit constraints
+	setImplicitConstraints(args.Job)
+
 	// Validate the job.
 	if err := validateJob(args.Job); err != nil {
 		return err
@@ -115,28 +118,6 @@ func (j *Job) Register(args *structs.JobRegisterRequest, reply *structs.JobRegis
 				}
 			}
 		}
-
-		// Add implicit constraints that the task groups are run on a Node with
-		// Vault
-		for _, tg := range args.Job.TaskGroups {
-			_, ok := policies[tg.Name]
-			if !ok {
-				// Not requesting Vault
-				continue
-			}
-
-			found := false
-			for _, c := range tg.Constraints {
-				if c.Equal(vaultConstraint) {
-					found = true
-					break
-				}
-			}
-
-			if !found {
-				tg.Constraints = append(tg.Constraints, vaultConstraint)
-			}
-		}
 	}
 
 	// Clear the Vault token
@@ -188,6 +169,77 @@ func (j *Job) Register(args *structs.JobRegisterRequest, reply *structs.JobRegis
 	return nil
 }
 
+// setImplicitConstraints adds implicit constraints to the job based on the
+// features it is requesting.
+func setImplicitConstraints(j *structs.Job) {
+	// Get the required Vault Policies
+	policies := j.VaultPolicies()
+
+	// Get the required signals
+	signals := j.RequiredSignals()
+
+	// Hot path
+	if len(signals) == 0 && len(policies) == 0 {
+		return
+	}
+
+	// Add Vault constraints
+	for _, tg := range j.TaskGroups {
+		_, ok := policies[tg.Name]
+		if !ok {
+			// Not requesting Vault
+			continue
+		}
+
+		found := false
+		for _, c := range tg.Constraints {
+			if c.Equal(vaultConstraint) {
+				found = true
+				break
+			}
+		}
+
+		if !found {
+			tg.Constraints = append(tg.Constraints, vaultConstraint)
+		}
+	}
+
+	// Add signal constraints
+	for _, tg := range j.TaskGroups {
+		tgSignals, ok := signals[tg.Name]
+		if !ok {
+			// Not requesting Vault
+			continue
+		}
+
+		// Flatten the signals
+		required := structs.MapStringStringSliceValueSet(tgSignals)
+		sigConstraint := getSignalConstraint(required)
+
+		found := false
+		for _, c := range tg.Constraints {
+			if c.Equal(sigConstraint) {
+				found = true
+				break
+			}
+		}
+
+		if !found {
+			tg.Constraints = append(tg.Constraints, sigConstraint)
+		}
+	}
+}
+
+// getSignalConstraint builds a suitable constraint based on the required
+// signals
+func getSignalConstraint(signals []string) *structs.Constraint {
+	return &structs.Constraint{
+		Operand: structs.ConstraintSetContains,
+		LTarget: "${attr.os.signals}",
+		RTarget: strings.Join(signals, ","),
+	}
+}
+
 // Summary retreives the summary of a job
 func (j *Job) Summary(args *structs.JobSummaryRequest,
 	reply *structs.JobSummaryResponse) error {
@@ -556,6 +608,9 @@ func (j *Job) Plan(args *structs.JobPlanRequest, reply *structs.JobPlanResponse)
 	// Initialize the job fields (sets defaults and any necessary init work).
 	args.Job.Canonicalize()
 
+	// Add implicit constraints
+	setImplicitConstraints(args.Job)
+
 	// Validate the job.
 	if err := validateJob(args.Job); err != nil {
 		return err
@@ -656,8 +711,14 @@ func validateJob(job *structs.Job) error {
 		multierror.Append(validationErrors, err)
 	}
 
+	// Get the signals required
+	signals := job.RequiredSignals()
+
 	// Validate the driver configurations.
 	for _, tg := range job.TaskGroups {
+		// Get the signals for the task group
+		tgSignals, tgOk := signals[tg.Name]
+
 		for _, task := range tg.Tasks {
 			d, err := driver.NewDriver(
 				task.Driver,
@@ -673,6 +734,21 @@ func validateJob(job *structs.Job) error {
 				formatted := fmt.Errorf("group %q -> task %q -> config: %v", tg.Name, task.Name, err)
 				multierror.Append(validationErrors, formatted)
 			}
+
+			// The task group didn't have any task that required signals
+			if !tgOk {
+				continue
+			}
+
+			// This task requires signals. Ensure the driver is capable
+			if required, ok := tgSignals[task.Name]; ok {
+				abilities := d.Abilities()
+				if !abilities.SendSignals {
+					formatted := fmt.Errorf("group %q -> task %q: driver %q doesn't support sending signals. Requested signals are %v",
+						tg.Name, task.Name, task.Driver, strings.Join(required, ", "))
+					multierror.Append(validationErrors, formatted)
+				}
+			}
 		}
 	}