CSI: make plugin health_timeout configurable in csi_plugin stanza

.changelog/13340.txt

@@ -0,0 +1,3 @@
+```release-note:improvements
+csi: Made the CSI Plugin supervisor health check configurable with a new CSI Stanza health_timeout field
+```

api/tasks.go

@@ -1039,10 +1039,18 @@ type TaskCSIPluginConfig struct {
 	//
 	// Default is /csi.
 	MountDir string `mapstructure:"mount_dir" hcl:"mount_dir,optional"`
+
+	// HealthTimeout is the time after which the CSI plugin tasks will be killed
+	// if the CSI Plugin is not healthy.
+	HealthTimeout time.Duration `mapstructure:"health_timeout" hcl:"health_timeout,optional"`
 }
 
 func (t *TaskCSIPluginConfig) Canonicalize() {
 	if t.MountDir == "" {
 		t.MountDir = "/csi"
 	}
+
+	if t.HealthTimeout == 0 {
+		t.HealthTimeout = 30 * time.Second
+	}
 }

client/allocrunner/taskrunner/plugin_supervisor_hook.go

@@ -253,7 +253,7 @@ func (h *csiPluginSupervisorHook) ensureSupervisorLoop(ctx context.Context) {
 
 	// We're in Poststart at this point, so if we can't connect within
 	// this deadline, assume it's broken so we can restart the task
-	startCtx, startCancelFn := context.WithTimeout(ctx, 30*time.Second)
+	startCtx, startCancelFn := context.WithTimeout(ctx, h.task.CSIPluginConfig.HealthTimeout)
 	defer startCancelFn()
 
 	var err error
@@ -441,7 +441,7 @@ func (h *csiPluginSupervisorHook) kill(ctx context.Context, reason error) {
 	if err := h.lifecycle.Kill(ctx,
 		structs.NewTaskEvent(structs.TaskKilling).
 			SetFailsTask().
-			SetDisplayMessage("CSI plugin did not become healthy before timeout"),
+			SetDisplayMessage(fmt.Sprintf("CSI plugin did not become healthy before configured %v health timeout", h.task.CSIPluginConfig.HealthTimeout.String())),
 	); err != nil {
 		h.logger.Error("failed to kill task", "kill_reason", reason, "error", err)
 	}

command/agent/job_endpoint.go

@@ -1263,6 +1263,7 @@ func ApiCSIPluginConfigToStructsCSIPluginConfig(apiConfig *api.TaskCSIPluginConf
 	sc.ID = apiConfig.ID
 	sc.Type = structs.CSIPluginType(apiConfig.Type)
 	sc.MountDir = apiConfig.MountDir
+	sc.HealthTimeout = apiConfig.HealthTimeout
 	return sc
 }
 

jobspec/parse_task.go

@@ -158,12 +158,20 @@ func parseTask(item *ast.ObjectItem, keys []string) (*api.Task, error) {
 		i := o.Elem().Items[0]
 
 		var m map[string]interface{}
+		var cfg api.TaskCSIPluginConfig
 		if err := hcl.DecodeObject(&m, i.Val); err != nil {
 			return nil, err
 		}
 
-		var cfg api.TaskCSIPluginConfig
-		if err := mapstructure.WeakDecode(m, &cfg); err != nil {
+		dec, err := mapstructure.NewDecoder(&mapstructure.DecoderConfig{
+			DecodeHook:       mapstructure.StringToTimeDurationHookFunc(),
+			WeaklyTypedInput: true,
+			Result:           &cfg,
+		})
+		if err != nil {
+			return nil, err
+		}
+		if err := dec.Decode(m); err != nil {
 			return nil, err
 		}
 

jobspec/parse_test.go

@@ -626,9 +626,10 @@ func TestParse(t *testing.T) {
 								Name:   "binstore",
 								Driver: "docker",
 								CSIPluginConfig: &api.TaskCSIPluginConfig{
-									ID:       "org.hashicorp.csi",
-									Type:     api.CSIPluginTypeMonolith,
-									MountDir: "/csi/test",
+									ID:            "org.hashicorp.csi",
+									Type:          api.CSIPluginTypeMonolith,
+									MountDir:      "/csi/test",
+									HealthTimeout: 1 * time.Minute,
 								},
 							},
 						},

jobspec/test-fixtures/csi-plugin.hcl

@@ -4,9 +4,10 @@ job "binstore-storagelocker" {
       driver = "docker"
 
       csi_plugin {
-        id        = "org.hashicorp.csi"
-        type      = "monolith"
-        mount_dir = "/csi/test"
+        id             = "org.hashicorp.csi"
+        type           = "monolith"
+        mount_dir      = "/csi/test"
+        health_timeout = "1m"
       }
     }
   }

nomad/structs/csi.go

@@ -67,6 +67,10 @@ type TaskCSIPluginConfig struct {
 	// to be created by the plugin, and will provide references into
 	// "MountDir/CSIIntermediaryDirname/{VolumeName}/{AllocID} for mounts.
 	MountDir string
+
+	// HealthTimeout is the time after which the CSI plugin tasks will be killed
+	// if the CSI Plugin is not healthy.
+	HealthTimeout time.Duration `mapstructure:"health_timeout" hcl:"health_timeout,optional"`
 }
 
 func (t *TaskCSIPluginConfig) Copy() *TaskCSIPluginConfig {

nomad/structs/structs.go

@@ -7255,6 +7255,10 @@ func (t *Task) Validate(ephemeralDisk *EphemeralDisk, jobType string, tgServices
 			mErr.Errors = append(mErr.Errors, fmt.Errorf("CSIPluginConfig PluginType must be one of 'node', 'controller', or 'monolith', got: \"%s\"", t.CSIPluginConfig.Type))
 		}
 
+		if t.CSIPluginConfig.HealthTimeout == 0 {
+			t.CSIPluginConfig.HealthTimeout = 30 * time.Second
+		}
+
 		// TODO: Investigate validation of the PluginMountDir. Not much we can do apart from check IsAbs until after we understand its execution environment though :(
 	}