periodic GC for CSI plugins

nomad/config.go

@@ -191,6 +191,13 @@ type Config struct {
 	// for GC. This gives users some time to view terminal deployments.
 	DeploymentGCThreshold time.Duration
 
+	// CSIPluginGCInterval is how often we dispatch a job to GC unused plugins.
+	CSIPluginGCInterval time.Duration
+
+	// CSIPluginGCThreshold is how "old" a plugin must be to be eligible
+	// for GC. This gives users some time to debug plugins.
+	CSIPluginGCThreshold time.Duration
+
 	// EvalNackTimeout controls how long we allow a sub-scheduler to
 	// work on an evaluation before we consider it failed and Nack it.
 	// This allows that evaluation to be handed to another sub-scheduler
@@ -377,6 +384,8 @@ func DefaultConfig() *Config {
 		NodeGCThreshold:                  24 * time.Hour,
 		DeploymentGCInterval:             5 * time.Minute,
 		DeploymentGCThreshold:            1 * time.Hour,
+		CSIPluginGCInterval:              5 * time.Minute,
+		CSIPluginGCThreshold:             1 * time.Hour,
 		EvalNackTimeout:                  60 * time.Second,
 		EvalDeliveryLimit:                3,
 		EvalNackInitialReenqueueDelay:    1 * time.Second,

nomad/core_sched.go

@@ -54,6 +54,8 @@ func (c *CoreScheduler) Process(eval *structs.Evaluation) error {
 		return c.deploymentGC(eval)
 	case structs.CoreJobCSIVolumeClaimGC:
 		return c.csiVolumeClaimGC(eval)
+	case structs.CoreJobCSIPluginGC:
+		return c.csiPluginGC(eval)
 	case structs.CoreJobForceGC:
 		return c.forceGC(eval)
 	default:
@@ -736,3 +738,9 @@ func (c *CoreScheduler) csiVolumeClaimGC(eval *structs.Evaluation) error {
 	err := c.srv.RPC("CSIVolume.Claim", req, &structs.CSIVolumeClaimResponse{})
 	return err
 }
+
+// csiPluginGC is used to garbage collect unused plugins
+func (c *CoreScheduler) csiPluginGC(eval *structs.Evaluation) error {
+	c.logger.Trace("garbage collecting unused CSI plugins")
+	return nil
+}

nomad/leader.go

@@ -522,6 +522,8 @@ func (s *Server) schedulePeriodic(stopCh chan struct{}) {
 	defer jobGC.Stop()
 	deploymentGC := time.NewTicker(s.config.DeploymentGCInterval)
 	defer deploymentGC.Stop()
+	csiPluginGC := time.NewTicker(s.config.CSIPluginGCInterval)
+	defer csiPluginGC.Stop()
 
 	// getLatest grabs the latest index from the state store. It returns true if
 	// the index was retrieved successfully.
@@ -554,6 +556,11 @@ func (s *Server) schedulePeriodic(stopCh chan struct{}) {
 			if index, ok := getLatest(); ok {
 				s.evalBroker.Enqueue(s.coreJobEval(structs.CoreJobDeploymentGC, index))
 			}
+		case <-csiPluginGC.C:
+			if index, ok := getLatest(); ok {
+				s.evalBroker.Enqueue(s.coreJobEval(structs.CoreJobCSIPluginGC, index))
+			}
+
 		case <-stopCh:
 			return
 		}

nomad/state/state_store.go

@@ -2312,6 +2312,34 @@ func (s *StateStore) CSIPluginDenormalize(ws memdb.WatchSet, plug *structs.CSIPl
 	return plug, nil
 }
 
+// UpsertCSIPlugin writes the plugin to the state store. Note: there
+// is currently no raft message for this, as it's intended to support
+// testing use cases.
+func (s *StateStore) UpsertCSIPlugin(index uint64, plug *structs.CSIPlugin) error {
+	txn := s.db.Txn(true)
+	defer txn.Abort()
+
+	existing, err := txn.First("csi_plugins", "id", plug.ID)
+	if err != nil {
+		return fmt.Errorf("csi_plugin lookup error: %s %v", plug.ID, err)
+	}
+
+	plug.ModifyIndex = index
+	if existing != nil {
+		plug.CreateIndex = existing.(*structs.CSIPlugin).CreateIndex
+	}
+
+	err = txn.Insert("csi_plugins", plug)
+	if err != nil {
+		return fmt.Errorf("csi_plugins insert error: %v", err)
+	}
+	if err := txn.Insert("index", &IndexEntry{"csi_plugins", index}); err != nil {
+		return fmt.Errorf("index update failed: %v", err)
+	}
+	txn.Commit()
+	return nil
+}
+
 // UpsertPeriodicLaunch is used to register a launch or update it.
 func (s *StateStore) UpsertPeriodicLaunch(index uint64, launch *structs.PeriodicLaunch) error {
 	txn := s.db.Txn(true)

nomad/structs/structs.go

@@ -8990,6 +8990,11 @@ const (
 	// claiming them. If so, we unclaim the volume.
 	CoreJobCSIVolumeClaimGC = "csi-volume-claim-gc"
 
+	// CoreJobCSIPluginGC is use for the garbage collection of CSI plugins.
+	// We periodically scan plugins to see if they have no associated volumes
+	// or allocs running them. If so, we delete the plugin.
+	CoreJobCSIPluginGC = "csi-plugin-gc"
+
 	// CoreJobForceGC is used to force garbage collection of all GCable objects.
 	CoreJobForceGC = "force-gc"
 )