a7a64443e1
This changeset adds a subsystem to run on the leader, similar to the deployment watcher or node drainer. The `Watcher` performs a blocking query on updates to the `CSIVolumes` table and triggers reaping of volume claims. This will avoid tying up scheduling workers by immediately sending volume claim workloads into their own loop, rather than blocking the scheduling workers in the core GC job doing things like talking to CSI controllers The volume watcher is enabled on leader step-up and disabled on leader step-down. The volume claim GC mechanism now makes an empty claim RPC for the volume to trigger an index bump. That in turn unblocks the blocking query in the volume watcher so it can assess which claims can be released for a volume.
32 lines
1.1 KiB
Go
32 lines
1.1 KiB
Go
package nomad
|
|
|
|
import (
|
|
"github.com/hashicorp/nomad/nomad/structs"
|
|
)
|
|
|
|
// volumeWatcherRaftShim is the shim that provides the state watching
|
|
// methods. These should be set by the server and passed to the volume
|
|
// watcher.
|
|
type volumeWatcherRaftShim struct {
|
|
// apply is used to apply a message to Raft
|
|
apply raftApplyFn
|
|
}
|
|
|
|
// convertApplyErrors parses the results of a raftApply and returns the index at
|
|
// which it was applied and any error that occurred. Raft Apply returns two
|
|
// separate errors, Raft library errors and user returned errors from the FSM.
|
|
// This helper, joins the errors by inspecting the applyResponse for an error.
|
|
func (shim *volumeWatcherRaftShim) convertApplyErrors(applyResp interface{}, index uint64, err error) (uint64, error) {
|
|
if applyResp != nil {
|
|
if fsmErr, ok := applyResp.(error); ok && fsmErr != nil {
|
|
return index, fsmErr
|
|
}
|
|
}
|
|
return index, err
|
|
}
|
|
|
|
func (shim *volumeWatcherRaftShim) UpsertVolumeClaims(req *structs.CSIVolumeClaimBatchRequest) (uint64, error) {
|
|
fsmErrIntf, index, raftErr := shim.apply(structs.CSIVolumeClaimBatchRequestType, req)
|
|
return shim.convertApplyErrors(fsmErrIntf, index, raftErr)
|
|
}
|