Track plan rejection history and automatically mark clients as inelig…

…ible (#13421)

Plan rejections occur when the scheduler work and the leader plan
applier disagree on the feasibility of a plan. This may happen for valid
reasons: since Nomad does parallel scheduling, it is expected that
different workers will have a different state when computing placements.

As the final plan reaches the leader plan applier, it may no longer be
valid due to a concurrent scheduling taking up intended resources. In
these situations the plan applier will notify the worker that the plan
was rejected and that they should refresh their state before trying
again.

In some rare and unexpected circumstances it has been observed that
workers will repeatedly submit the same plan, even if they are always
rejected.

While the root cause is still unknown this mitigation has been put in
place. The plan applier will now track the history of plan rejections
per client and include in the plan result a list of node IDs that should
be set as ineligible if the number of rejections in a given time window
crosses a certain threshold. The window size and threshold value can be
adjusted in the server configuration.

To avoid marking several nodes as ineligible at one, the operation is rate
limited to 5 nodes every 30min, with an initial burst of 10 operations.

.changelog/13421.txt

@@ -0,0 +1,7 @@
+```release-note:improvement
+core: automatically mark clients with recurring plan rejections as ineligible
+```
+
+```release-note:improvement
+metrics: emit `nomad.nomad.plan.rejection_tracker.node_score` metric for the number of times a node had a plan rejection within the past time window
+```

command/agent/agent.go

@@ -439,6 +439,20 @@ func convertServerConfig(agentConfig *Config) (*nomad.Config, error) {
 		return nil, fmt.Errorf("deploy_query_rate_limit must be greater than 0")
 	}
 
+	// Set plan rejection tracker configuration.
+	if planRejectConf := agentConfig.Server.PlanRejectionTracker; planRejectConf != nil {
+		if planRejectConf.Enabled != nil {
+			conf.NodePlanRejectionEnabled = *planRejectConf.Enabled
+		}
+		conf.NodePlanRejectionThreshold = planRejectConf.NodeThreshold
+
+		if planRejectConf.NodeWindow == 0 {
+			return nil, fmt.Errorf("plan_rejection_tracker.node_window must be greater than 0")
+		} else {
+			conf.NodePlanRejectionWindow = planRejectConf.NodeWindow
+		}
+	}
+
 	// Add Enterprise license configs
 	conf.LicenseEnv = agentConfig.Server.LicenseEnv
 	conf.LicensePath = agentConfig.Server.LicensePath

command/agent/agent_test.go

@@ -366,6 +366,82 @@ func TestAgent_ServerConfig_Limits_OK(t *testing.T) {
 	}
 }
 
+func TestAgent_ServerConfig_PlanRejectionTracker(t *testing.T) {
+	ci.Parallel(t)
+
+	cases := []struct {
+		name           string
+		trackerConfig  *PlanRejectionTracker
+		expectedConfig *PlanRejectionTracker
+		expectedErr    string
+	}{
+		{
+			name:          "default",
+			trackerConfig: nil,
+			expectedConfig: &PlanRejectionTracker{
+				NodeThreshold: 100,
+				NodeWindow:    5 * time.Minute,
+			},
+			expectedErr: "",
+		},
+		{
+			name: "valid config",
+			trackerConfig: &PlanRejectionTracker{
+				Enabled:       helper.BoolToPtr(true),
+				NodeThreshold: 123,
+				NodeWindow:    17 * time.Minute,
+			},
+			expectedConfig: &PlanRejectionTracker{
+				Enabled:       helper.BoolToPtr(true),
+				NodeThreshold: 123,
+				NodeWindow:    17 * time.Minute,
+			},
+			expectedErr: "",
+		},
+		{
+			name: "invalid node window",
+			trackerConfig: &PlanRejectionTracker{
+				NodeThreshold: 123,
+			},
+			expectedErr: "node_window must be greater than 0",
+		},
+	}
+
+	for _, tc := range cases {
+		t.Run(tc.name, func(t *testing.T) {
+			config := DevConfig(nil)
+			require.NoError(t, config.normalizeAddrs())
+
+			if tc.trackerConfig != nil {
+				config.Server.PlanRejectionTracker = tc.trackerConfig
+			}
+
+			serverConfig, err := convertServerConfig(config)
+
+			if tc.expectedErr != "" {
+				require.Error(t, err)
+				require.Contains(t, err.Error(), tc.expectedErr)
+			} else {
+				require.NoError(t, err)
+				if tc.expectedConfig.Enabled != nil {
+					require.Equal(t,
+						*tc.expectedConfig.Enabled,
+						serverConfig.NodePlanRejectionEnabled,
+					)
+				}
+				require.Equal(t,
+					tc.expectedConfig.NodeThreshold,
+					serverConfig.NodePlanRejectionThreshold,
+				)
+				require.Equal(t,
+					tc.expectedConfig.NodeWindow,
+					serverConfig.NodePlanRejectionWindow,
+				)
+			}
+		})
+	}
+}
+
 func TestAgent_ServerConfig_RaftMultiplier_Ok(t *testing.T) {
 	ci.Parallel(t)
 

command/agent/config.go

@@ -516,6 +516,10 @@ type ServerConfig struct {
 	// This value is ignored.
 	DefaultSchedulerConfig *structs.SchedulerConfiguration `hcl:"default_scheduler_config"`
 
+	// PlanRejectionTracker configures the node plan rejection tracker that
+	// detects potentially bad nodes.
+	PlanRejectionTracker *PlanRejectionTracker `hcl:"plan_rejection_tracker"`
+
 	// EnableEventBroker configures whether this server's state store
 	// will generate events for its event stream.
 	EnableEventBroker *bool `hcl:"enable_event_broker"`
@@ -561,6 +565,53 @@ type RaftBoltConfig struct {
 	NoFreelistSync bool `hcl:"no_freelist_sync"`
 }
 
+// PlanRejectionTracker is used in servers to configure the plan rejection
+// tracker.
+type PlanRejectionTracker struct {
+	// Enabled controls if the plan rejection tracker is active or not.
+	Enabled *bool `hcl:"enabled"`
+
+	// NodeThreshold is the number of times a node can have plan rejections
+	// before it is marked as ineligible.
+	NodeThreshold int `hcl:"node_threshold"`
+
+	// NodeWindow is the time window used to track active plan rejections for
+	// nodes.
+	NodeWindow    time.Duration
+	NodeWindowHCL string `hcl:"node_window" json:"-"`
+
+	// ExtraKeysHCL is used by hcl to surface unexpected keys
+	ExtraKeysHCL []string `hcl:",unusedKeys" json:"-"`
+}
+
+func (p *PlanRejectionTracker) Merge(b *PlanRejectionTracker) *PlanRejectionTracker {
+	if p == nil {
+		return b
+	}
+
+	result := *p
+
+	if b == nil {
+		return &result
+	}
+
+	if b.Enabled != nil {
+		result.Enabled = b.Enabled
+	}
+
+	if b.NodeThreshold != 0 {
+		result.NodeThreshold = b.NodeThreshold
+	}
+
+	if b.NodeWindow != 0 {
+		result.NodeWindow = b.NodeWindow
+	}
+	if b.NodeWindowHCL != "" {
+		result.NodeWindowHCL = b.NodeWindowHCL
+	}
+	return &result
+}
+
 // Search is used in servers to configure search API options.
 type Search struct {
 	// FuzzyEnabled toggles whether the FuzzySearch API is enabled. If not
@@ -998,6 +1049,11 @@ func DefaultConfig() *Config {
 			EventBufferSize:   helper.IntToPtr(100),
 			RaftProtocol:      3,
 			StartJoin:         []string{},
+			PlanRejectionTracker: &PlanRejectionTracker{
+				Enabled:       helper.BoolToPtr(false),
+				NodeThreshold: 100,
+				NodeWindow:    5 * time.Minute,
+			},
 			ServerJoin: &ServerJoin{
 				RetryJoin:        []string{},
 				RetryInterval:    30 * time.Second,
@@ -1608,6 +1664,10 @@ func (s *ServerConfig) Merge(b *ServerConfig) *ServerConfig {
 		result.EventBufferSize = b.EventBufferSize
 	}
 
+	if b.PlanRejectionTracker != nil {
+		result.PlanRejectionTracker = result.PlanRejectionTracker.Merge(b.PlanRejectionTracker)
+	}
+
 	if b.DefaultSchedulerConfig != nil {
 		c := *b.DefaultSchedulerConfig
 		result.DefaultSchedulerConfig = &c

command/agent/config_parse.go

@@ -43,9 +43,12 @@ func ParseConfigFile(path string) (*Config, error) {
 				VaultRetry:  &client.RetryConfig{},
 			},
 		},
+		Server: &ServerConfig{
+			PlanRejectionTracker: &PlanRejectionTracker{},
+			ServerJoin:           &ServerJoin{},
+		},
 		ACL:       &ACLConfig{},
 		Audit:     &config.AuditConfig{},
-		Server:    &ServerConfig{ServerJoin: &ServerJoin{}},
 		Consul:    &config.ConsulConfig{},
 		Autopilot: &config.AutopilotConfig{},
 		Telemetry: &Telemetry{},
@@ -54,7 +57,7 @@ func ParseConfigFile(path string) (*Config, error) {
 
 	err = hcl.Decode(c, buf.String())
 	if err != nil {
-		return nil, err
+		return nil, fmt.Errorf("failed to decode HCL file %s: %w", path, err)
 	}
 
 	// convert strings to time.Durations
@@ -66,6 +69,7 @@ func ParseConfigFile(path string) (*Config, error) {
 		{"server.heartbeat_grace", &c.Server.HeartbeatGrace, &c.Server.HeartbeatGraceHCL, nil},
 		{"server.min_heartbeat_ttl", &c.Server.MinHeartbeatTTL, &c.Server.MinHeartbeatTTLHCL, nil},
 		{"server.failover_heartbeat_ttl", &c.Server.FailoverHeartbeatTTL, &c.Server.FailoverHeartbeatTTLHCL, nil},
+		{"server.plan_rejection_tracker.node_window", &c.Server.PlanRejectionTracker.NodeWindow, &c.Server.PlanRejectionTracker.NodeWindowHCL, nil},
 		{"server.retry_interval", &c.Server.RetryInterval, &c.Server.RetryIntervalHCL, nil},
 		{"server.server_join.retry_interval", &c.Server.ServerJoin.RetryInterval, &c.Server.ServerJoin.RetryIntervalHCL, nil},
 		{"consul.timeout", &c.Consul.Timeout, &c.Consul.TimeoutHCL, nil},

command/agent/config_parse_test.go

@@ -126,6 +126,12 @@ var basicConfig = &Config{
 		EncryptKey:                "abc",
 		EnableEventBroker:         helper.BoolToPtr(false),
 		EventBufferSize:           helper.IntToPtr(200),
+		PlanRejectionTracker: &PlanRejectionTracker{
+			Enabled:       helper.BoolToPtr(true),
+			NodeThreshold: 100,
+			NodeWindow:    41 * time.Minute,
+			NodeWindowHCL: "41m",
+		},
 		ServerJoin: &ServerJoin{
 			RetryJoin:        []string{"1.1.1.1", "2.2.2.2"},
 			RetryInterval:    time.Duration(15) * time.Second,
@@ -543,6 +549,9 @@ func (c *Config) addDefaults() {
 	if c.Server.ServerJoin == nil {
 		c.Server.ServerJoin = &ServerJoin{}
 	}
+	if c.Server.PlanRejectionTracker == nil {
+		c.Server.PlanRejectionTracker = &PlanRejectionTracker{}
+	}
 }
 
 // Tests for a panic parsing json with an object of exactly
@@ -620,6 +629,11 @@ var sample0 = &Config{
 		RetryJoin:       []string{"10.0.0.101", "10.0.0.102", "10.0.0.103"},
 		EncryptKey:      "sHck3WL6cxuhuY7Mso9BHA==",
 		ServerJoin:      &ServerJoin{},
+		PlanRejectionTracker: &PlanRejectionTracker{
+			NodeThreshold: 100,
+			NodeWindow:    31 * time.Minute,
+			NodeWindowHCL: "31m",
+		},
 	},
 	ACL: &ACLConfig{
 		Enabled: true,
@@ -710,6 +724,11 @@ var sample1 = &Config{
 		RetryJoin:       []string{"10.0.0.101", "10.0.0.102", "10.0.0.103"},
 		EncryptKey:      "sHck3WL6cxuhuY7Mso9BHA==",
 		ServerJoin:      &ServerJoin{},
+		PlanRejectionTracker: &PlanRejectionTracker{
+			NodeThreshold: 100,
+			NodeWindow:    31 * time.Minute,
+			NodeWindowHCL: "31m",
+		},
 	},
 	ACL: &ACLConfig{
 		Enabled: true,

command/agent/config_test.go

@@ -148,6 +148,11 @@ func TestConfig_Merge(t *testing.T) {
 			UpgradeVersion:         "foo",
 			EnableEventBroker:      helper.BoolToPtr(false),
 			EventBufferSize:        helper.IntToPtr(0),
+			PlanRejectionTracker: &PlanRejectionTracker{
+				Enabled:       helper.BoolToPtr(true),
+				NodeThreshold: 100,
+				NodeWindow:    11 * time.Minute,
+			},
 		},
 		ACL: &ACLConfig{
 			Enabled:          true,
@@ -343,6 +348,11 @@ func TestConfig_Merge(t *testing.T) {
 			UpgradeVersion:         "bar",
 			EnableEventBroker:      helper.BoolToPtr(true),
 			EventBufferSize:        helper.IntToPtr(100),
+			PlanRejectionTracker: &PlanRejectionTracker{
+				Enabled:       helper.BoolToPtr(true),
+				NodeThreshold: 100,
+				NodeWindow:    11 * time.Minute,
+			},
 		},
 		ACL: &ACLConfig{
 			Enabled:          true,

command/agent/testdata/basic.hcl

@@ -133,6 +133,12 @@ server {
   enable_event_broker           = false
   event_buffer_size             = 200
 
+  plan_rejection_tracker {
+    enabled        = true
+    node_threshold = 100
+    node_window    = "41m"
+  }
+
   server_join {
     retry_join     = ["1.1.1.1", "2.2.2.2"]
     retry_max      = 3

command/agent/testdata/basic.json

@@ -278,6 +278,11 @@
       "node_gc_threshold": "12h",
       "non_voting_server": true,
       "num_schedulers": 2,
+      "plan_rejection_tracker": {
+        "enabled": true,
+        "node_threshold": 100,
+        "node_window": "41m"
+      },
       "raft_protocol": 3,
       "raft_multiplier": 4,
       "redundancy_zone": "foo",

command/agent/testdata/sample0.json

@@ -55,6 +55,10 @@
     "bootstrap_expect": 3,
     "enabled": true,
     "encrypt": "sHck3WL6cxuhuY7Mso9BHA==",
+    "plan_rejection_tracker": {
+      "node_threshold": 100,
+      "node_window": "31m"
+    },
     "retry_join": [
       "10.0.0.101",
       "10.0.0.102",

command/agent/testdata/sample1/sample0.json

@@ -20,6 +20,10 @@
     "bootstrap_expect": 3,
     "enabled": true,
     "encrypt": "sHck3WL6cxuhuY7Mso9BHA==",
+    "plan_rejection_tracker": {
+      "node_threshold": 100,
+      "node_window": "31m"
+    },
     "retry_join": [
       "10.0.0.101",
       "10.0.0.102",

nomad/config.go

@@ -248,6 +248,17 @@ type Config struct {
 	// additional delay is selected from this range randomly.
 	EvalFailedFollowupDelayRange time.Duration
 
+	// NodePlanRejectionEnabled controls if node rejection tracker is enabled.
+	NodePlanRejectionEnabled bool
+
+	// NodePlanRejectionThreshold is the number of times a node must have a
+	// plan rejection before it is set as ineligible.
+	NodePlanRejectionThreshold int
+
+	// NodePlanRejectionWindow is the time window used to track plan
+	// rejections for nodes.
+	NodePlanRejectionWindow time.Duration
+
 	// MinHeartbeatTTL is the minimum time between heartbeats.
 	// This is used as a floor to prevent excessive updates.
 	MinHeartbeatTTL time.Duration
@@ -415,6 +426,9 @@ func DefaultConfig() *Config {
 		MaxHeartbeatsPerSecond:           50.0,
 		HeartbeatGrace:                   10 * time.Second,
 		FailoverHeartbeatTTL:             300 * time.Second,
+		NodePlanRejectionEnabled:         false,
+		NodePlanRejectionThreshold:       15,
+		NodePlanRejectionWindow:          10 * time.Minute,
 		ConsulConfig:                     config.DefaultConsulConfig(),
 		VaultConfig:                      config.DefaultVaultConfig(),
 		RPCHoldTimeout:                   5 * time.Second,