actual_state_of_world.go 30 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876
  1. /*
  2. Copyright 2016 The Kubernetes Authors.
  3. Licensed under the Apache License, Version 2.0 (the "License");
  4. you may not use this file except in compliance with the License.
  5. You may obtain a copy of the License at
  6. http://www.apache.org/licenses/LICENSE-2.0
  7. Unless required by applicable law or agreed to in writing, software
  8. distributed under the License is distributed on an "AS IS" BASIS,
  9. WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  10. See the License for the specific language governing permissions and
  11. limitations under the License.
  12. */
  13. /*
  14. Package cache implements data structures used by the kubelet volume manager to
  15. keep track of attached volumes and the pods that mounted them.
  16. */
  17. package cache
  18. import (
  19. "fmt"
  20. "sync"
  21. "k8s.io/api/core/v1"
  22. "k8s.io/apimachinery/pkg/types"
  23. utilfeature "k8s.io/apiserver/pkg/util/feature"
  24. "k8s.io/klog"
  25. "k8s.io/kubernetes/pkg/features"
  26. "k8s.io/kubernetes/pkg/volume"
  27. "k8s.io/kubernetes/pkg/volume/util"
  28. "k8s.io/kubernetes/pkg/volume/util/operationexecutor"
  29. volumetypes "k8s.io/kubernetes/pkg/volume/util/types"
  30. )
  31. // ActualStateOfWorld defines a set of thread-safe operations for the kubelet
  32. // volume manager's actual state of the world cache.
  33. // This cache contains volumes->pods i.e. a set of all volumes attached to this
  34. // node and the pods that the manager believes have successfully mounted the
  35. // volume.
  36. // Note: This is distinct from the ActualStateOfWorld implemented by the
  37. // attach/detach controller. They both keep track of different objects. This
  38. // contains kubelet volume manager specific state.
  39. type ActualStateOfWorld interface {
  40. // ActualStateOfWorld must implement the methods required to allow
  41. // operationexecutor to interact with it.
  42. operationexecutor.ActualStateOfWorldMounterUpdater
  43. // ActualStateOfWorld must implement the methods required to allow
  44. // operationexecutor to interact with it.
  45. operationexecutor.ActualStateOfWorldAttacherUpdater
  46. // AddPodToVolume adds the given pod to the given volume in the cache
  47. // indicating the specified volume has been successfully mounted to the
  48. // specified pod.
  49. // If a pod with the same unique name already exists under the specified
  50. // volume, reset the pod's remountRequired value.
  51. // If a volume with the name volumeName does not exist in the list of
  52. // attached volumes, an error is returned.
  53. AddPodToVolume(podName volumetypes.UniquePodName, podUID types.UID, volumeName v1.UniqueVolumeName, mounter volume.Mounter, blockVolumeMapper volume.BlockVolumeMapper, outerVolumeSpecName string, volumeGidValue string, volumeSpec *volume.Spec) error
  54. // MarkRemountRequired marks each volume that is successfully attached and
  55. // mounted for the specified pod as requiring remount (if the plugin for the
  56. // volume indicates it requires remounting on pod updates). Atomically
  57. // updating volumes depend on this to update the contents of the volume on
  58. // pod update.
  59. MarkRemountRequired(podName volumetypes.UniquePodName)
  60. // SetVolumeGloballyMounted sets the GloballyMounted value for the given
  61. // volume. When set to true this value indicates that the volume is mounted
  62. // to the underlying device at a global mount point. This global mount point
  63. // must unmounted prior to detach.
  64. // If a volume with the name volumeName does not exist in the list of
  65. // attached volumes, an error is returned.
  66. SetVolumeGloballyMounted(volumeName v1.UniqueVolumeName, globallyMounted bool, devicePath, deviceMountPath string) error
  67. // DeletePodFromVolume removes the given pod from the given volume in the
  68. // cache indicating the volume has been successfully unmounted from the pod.
  69. // If a pod with the same unique name does not exist under the specified
  70. // volume, this is a no-op.
  71. // If a volume with the name volumeName does not exist in the list of
  72. // attached volumes, an error is returned.
  73. DeletePodFromVolume(podName volumetypes.UniquePodName, volumeName v1.UniqueVolumeName) error
  74. // DeleteVolume removes the given volume from the list of attached volumes
  75. // in the cache indicating the volume has been successfully detached from
  76. // this node.
  77. // If a volume with the name volumeName does not exist in the list of
  78. // attached volumes, this is a no-op.
  79. // If a volume with the name volumeName exists and its list of mountedPods
  80. // is not empty, an error is returned.
  81. DeleteVolume(volumeName v1.UniqueVolumeName) error
  82. // PodExistsInVolume returns true if the given pod exists in the list of
  83. // mountedPods for the given volume in the cache, indicating that the volume
  84. // is attached to this node and the pod has successfully mounted it.
  85. // If a pod with the same unique name does not exist under the specified
  86. // volume, false is returned.
  87. // If a volume with the name volumeName does not exist in the list of
  88. // attached volumes, a volumeNotAttachedError is returned indicating the
  89. // given volume is not yet attached.
  90. // If the given volumeName/podName combo exists but the value of
  91. // remountRequired is true, a remountRequiredError is returned indicating
  92. // the given volume has been successfully mounted to this pod but should be
  93. // remounted to reflect changes in the referencing pod. Atomically updating
  94. // volumes, depend on this to update the contents of the volume.
  95. // All volume mounting calls should be idempotent so a second mount call for
  96. // volumes that do not need to update contents should not fail.
  97. PodExistsInVolume(podName volumetypes.UniquePodName, volumeName v1.UniqueVolumeName) (bool, string, error)
  98. // VolumeExistsWithSpecName returns true if the given volume specified with the
  99. // volume spec name (a.k.a., InnerVolumeSpecName) exists in the list of
  100. // volumes that should be attached to this node.
  101. // If a pod with the same name does not exist under the specified
  102. // volume, false is returned.
  103. VolumeExistsWithSpecName(podName volumetypes.UniquePodName, volumeSpecName string) bool
  104. // VolumeExists returns true if the given volume exists in the list of
  105. // attached volumes in the cache, indicating the volume is attached to this
  106. // node.
  107. VolumeExists(volumeName v1.UniqueVolumeName) bool
  108. // GetMountedVolumes generates and returns a list of volumes and the pods
  109. // they are successfully attached and mounted for based on the current
  110. // actual state of the world.
  111. GetMountedVolumes() []MountedVolume
  112. // GetMountedVolumesForPod generates and returns a list of volumes that are
  113. // successfully attached and mounted for the specified pod based on the
  114. // current actual state of the world.
  115. GetMountedVolumesForPod(podName volumetypes.UniquePodName) []MountedVolume
  116. // GetGloballyMountedVolumes generates and returns a list of all attached
  117. // volumes that are globally mounted. This list can be used to determine
  118. // which volumes should be reported as "in use" in the node's VolumesInUse
  119. // status field. Globally mounted here refers to the shared plugin mount
  120. // point for the attachable volume from which the pod specific mount points
  121. // are created (via bind mount).
  122. GetGloballyMountedVolumes() []AttachedVolume
  123. // GetUnmountedVolumes generates and returns a list of attached volumes that
  124. // have no mountedPods. This list can be used to determine which volumes are
  125. // no longer referenced and may be globally unmounted and detached.
  126. GetUnmountedVolumes() []AttachedVolume
  127. // GetPods generates and returns a map of pods in which map is indexed
  128. // with pod's unique name. This map can be used to determine which pod is currently
  129. // in actual state of world.
  130. GetPods() map[volumetypes.UniquePodName]bool
  131. // MarkFSResizeRequired marks each volume that is successfully attached and
  132. // mounted for the specified pod as requiring file system resize (if the plugin for the
  133. // volume indicates it requires file system resize).
  134. MarkFSResizeRequired(volumeName v1.UniqueVolumeName, podName volumetypes.UniquePodName)
  135. // GetAttachedVolumes returns a list of volumes that is known to be attached
  136. // to the node. This list can be used to determine volumes that are either in-use
  137. // or have a mount/unmount operation pending.
  138. GetAttachedVolumes() []AttachedVolume
  139. }
  140. // MountedVolume represents a volume that has successfully been mounted to a pod.
  141. type MountedVolume struct {
  142. operationexecutor.MountedVolume
  143. }
  144. // AttachedVolume represents a volume that is attached to a node.
  145. type AttachedVolume struct {
  146. operationexecutor.AttachedVolume
  147. // GloballyMounted indicates that the volume is mounted to the underlying
  148. // device at a global mount point. This global mount point must unmounted
  149. // prior to detach.
  150. GloballyMounted bool
  151. }
  152. // NewActualStateOfWorld returns a new instance of ActualStateOfWorld.
  153. func NewActualStateOfWorld(
  154. nodeName types.NodeName,
  155. volumePluginMgr *volume.VolumePluginMgr) ActualStateOfWorld {
  156. return &actualStateOfWorld{
  157. nodeName: nodeName,
  158. attachedVolumes: make(map[v1.UniqueVolumeName]attachedVolume),
  159. volumePluginMgr: volumePluginMgr,
  160. }
  161. }
  162. // IsVolumeNotAttachedError returns true if the specified error is a
  163. // volumeNotAttachedError.
  164. func IsVolumeNotAttachedError(err error) bool {
  165. _, ok := err.(volumeNotAttachedError)
  166. return ok
  167. }
  168. // IsRemountRequiredError returns true if the specified error is a
  169. // remountRequiredError.
  170. func IsRemountRequiredError(err error) bool {
  171. _, ok := err.(remountRequiredError)
  172. return ok
  173. }
  174. type actualStateOfWorld struct {
  175. // nodeName is the name of this node. This value is passed to Attach/Detach
  176. nodeName types.NodeName
  177. // attachedVolumes is a map containing the set of volumes the kubelet volume
  178. // manager believes to be successfully attached to this node. Volume types
  179. // that do not implement an attacher interface are assumed to be in this
  180. // state by default.
  181. // The key in this map is the name of the volume and the value is an object
  182. // containing more information about the attached volume.
  183. attachedVolumes map[v1.UniqueVolumeName]attachedVolume
  184. // volumePluginMgr is the volume plugin manager used to create volume
  185. // plugin objects.
  186. volumePluginMgr *volume.VolumePluginMgr
  187. sync.RWMutex
  188. }
  189. // attachedVolume represents a volume the kubelet volume manager believes to be
  190. // successfully attached to a node it is managing. Volume types that do not
  191. // implement an attacher are assumed to be in this state.
  192. type attachedVolume struct {
  193. // volumeName contains the unique identifier for this volume.
  194. volumeName v1.UniqueVolumeName
  195. // mountedPods is a map containing the set of pods that this volume has been
  196. // successfully mounted to. The key in this map is the name of the pod and
  197. // the value is a mountedPod object containing more information about the
  198. // pod.
  199. mountedPods map[volumetypes.UniquePodName]mountedPod
  200. // spec is the volume spec containing the specification for this volume.
  201. // Used to generate the volume plugin object, and passed to plugin methods.
  202. // In particular, the Unmount method uses spec.Name() as the volumeSpecName
  203. // in the mount path:
  204. // /var/lib/kubelet/pods/{podUID}/volumes/{escapeQualifiedPluginName}/{volumeSpecName}/
  205. spec *volume.Spec
  206. // pluginName is the Unescaped Qualified name of the volume plugin used to
  207. // attach and mount this volume. It is stored separately in case the full
  208. // volume spec (everything except the name) can not be reconstructed for a
  209. // volume that should be unmounted (which would be the case for a mount path
  210. // read from disk without a full volume spec).
  211. pluginName string
  212. // pluginIsAttachable indicates the volume plugin used to attach and mount
  213. // this volume implements the volume.Attacher interface
  214. pluginIsAttachable bool
  215. // globallyMounted indicates that the volume is mounted to the underlying
  216. // device at a global mount point. This global mount point must be unmounted
  217. // prior to detach.
  218. globallyMounted bool
  219. // devicePath contains the path on the node where the volume is attached for
  220. // attachable volumes
  221. devicePath string
  222. // deviceMountPath contains the path on the node where the device should
  223. // be mounted after it is attached.
  224. deviceMountPath string
  225. }
  226. // The mountedPod object represents a pod for which the kubelet volume manager
  227. // believes the underlying volume has been successfully been mounted.
  228. type mountedPod struct {
  229. // the name of the pod
  230. podName volumetypes.UniquePodName
  231. // the UID of the pod
  232. podUID types.UID
  233. // mounter used to mount
  234. mounter volume.Mounter
  235. // mapper used to block volumes support
  236. blockVolumeMapper volume.BlockVolumeMapper
  237. // spec is the volume spec containing the specification for this volume.
  238. // Used to generate the volume plugin object, and passed to plugin methods.
  239. // In particular, the Unmount method uses spec.Name() as the volumeSpecName
  240. // in the mount path:
  241. // /var/lib/kubelet/pods/{podUID}/volumes/{escapeQualifiedPluginName}/{volumeSpecName}/
  242. volumeSpec *volume.Spec
  243. // outerVolumeSpecName is the volume.Spec.Name() of the volume as referenced
  244. // directly in the pod. If the volume was referenced through a persistent
  245. // volume claim, this contains the volume.Spec.Name() of the persistent
  246. // volume claim
  247. outerVolumeSpecName string
  248. // remountRequired indicates the underlying volume has been successfully
  249. // mounted to this pod but it should be remounted to reflect changes in the
  250. // referencing pod.
  251. // Atomically updating volumes depend on this to update the contents of the
  252. // volume. All volume mounting calls should be idempotent so a second mount
  253. // call for volumes that do not need to update contents should not fail.
  254. remountRequired bool
  255. // volumeGidValue contains the value of the GID annotation, if present.
  256. volumeGidValue string
  257. // fsResizeRequired indicates the underlying volume has been successfully
  258. // mounted to this pod but its size has been expanded after that.
  259. fsResizeRequired bool
  260. }
  261. func (asw *actualStateOfWorld) MarkVolumeAsAttached(
  262. volumeName v1.UniqueVolumeName, volumeSpec *volume.Spec, _ types.NodeName, devicePath string) error {
  263. return asw.addVolume(volumeName, volumeSpec, devicePath)
  264. }
  265. func (asw *actualStateOfWorld) MarkVolumeAsUncertain(
  266. volumeName v1.UniqueVolumeName, volumeSpec *volume.Spec, _ types.NodeName) error {
  267. return nil
  268. }
  269. func (asw *actualStateOfWorld) MarkVolumeAsDetached(
  270. volumeName v1.UniqueVolumeName, nodeName types.NodeName) {
  271. asw.DeleteVolume(volumeName)
  272. }
  273. func (asw *actualStateOfWorld) MarkVolumeAsMounted(
  274. podName volumetypes.UniquePodName,
  275. podUID types.UID,
  276. volumeName v1.UniqueVolumeName,
  277. mounter volume.Mounter,
  278. blockVolumeMapper volume.BlockVolumeMapper,
  279. outerVolumeSpecName string,
  280. volumeGidValue string,
  281. volumeSpec *volume.Spec) error {
  282. return asw.AddPodToVolume(
  283. podName,
  284. podUID,
  285. volumeName,
  286. mounter,
  287. blockVolumeMapper,
  288. outerVolumeSpecName,
  289. volumeGidValue,
  290. volumeSpec)
  291. }
  292. func (asw *actualStateOfWorld) AddVolumeToReportAsAttached(volumeName v1.UniqueVolumeName, nodeName types.NodeName) {
  293. // no operation for kubelet side
  294. }
  295. func (asw *actualStateOfWorld) RemoveVolumeFromReportAsAttached(volumeName v1.UniqueVolumeName, nodeName types.NodeName) error {
  296. // no operation for kubelet side
  297. return nil
  298. }
  299. func (asw *actualStateOfWorld) MarkVolumeAsUnmounted(
  300. podName volumetypes.UniquePodName, volumeName v1.UniqueVolumeName) error {
  301. return asw.DeletePodFromVolume(podName, volumeName)
  302. }
  303. func (asw *actualStateOfWorld) MarkDeviceAsMounted(
  304. volumeName v1.UniqueVolumeName, devicePath, deviceMountPath string) error {
  305. return asw.SetVolumeGloballyMounted(volumeName, true /* globallyMounted */, devicePath, deviceMountPath)
  306. }
  307. func (asw *actualStateOfWorld) MarkDeviceAsUnmounted(
  308. volumeName v1.UniqueVolumeName) error {
  309. return asw.SetVolumeGloballyMounted(volumeName, false /* globallyMounted */, "", "")
  310. }
  311. // addVolume adds the given volume to the cache indicating the specified
  312. // volume is attached to this node. If no volume name is supplied, a unique
  313. // volume name is generated from the volumeSpec and returned on success. If a
  314. // volume with the same generated name already exists, this is a noop. If no
  315. // volume plugin can support the given volumeSpec or more than one plugin can
  316. // support it, an error is returned.
  317. func (asw *actualStateOfWorld) addVolume(
  318. volumeName v1.UniqueVolumeName, volumeSpec *volume.Spec, devicePath string) error {
  319. asw.Lock()
  320. defer asw.Unlock()
  321. volumePlugin, err := asw.volumePluginMgr.FindPluginBySpec(volumeSpec)
  322. if err != nil || volumePlugin == nil {
  323. return fmt.Errorf(
  324. "failed to get Plugin from volumeSpec for volume %q err=%v",
  325. volumeSpec.Name(),
  326. err)
  327. }
  328. if len(volumeName) == 0 {
  329. volumeName, err = util.GetUniqueVolumeNameFromSpec(volumePlugin, volumeSpec)
  330. if err != nil {
  331. return fmt.Errorf(
  332. "failed to GetUniqueVolumeNameFromSpec for volumeSpec %q using volume plugin %q err=%v",
  333. volumeSpec.Name(),
  334. volumePlugin.GetPluginName(),
  335. err)
  336. }
  337. }
  338. pluginIsAttachable := false
  339. if _, ok := volumePlugin.(volume.AttachableVolumePlugin); ok {
  340. pluginIsAttachable = true
  341. }
  342. volumeObj, volumeExists := asw.attachedVolumes[volumeName]
  343. if !volumeExists {
  344. volumeObj = attachedVolume{
  345. volumeName: volumeName,
  346. spec: volumeSpec,
  347. mountedPods: make(map[volumetypes.UniquePodName]mountedPod),
  348. pluginName: volumePlugin.GetPluginName(),
  349. pluginIsAttachable: pluginIsAttachable,
  350. globallyMounted: false,
  351. devicePath: devicePath,
  352. }
  353. } else {
  354. // If volume object already exists, update the fields such as device path
  355. volumeObj.devicePath = devicePath
  356. klog.V(2).Infof("Volume %q is already added to attachedVolume list, update device path %q",
  357. volumeName,
  358. devicePath)
  359. }
  360. asw.attachedVolumes[volumeName] = volumeObj
  361. return nil
  362. }
  363. func (asw *actualStateOfWorld) AddPodToVolume(
  364. podName volumetypes.UniquePodName,
  365. podUID types.UID,
  366. volumeName v1.UniqueVolumeName,
  367. mounter volume.Mounter,
  368. blockVolumeMapper volume.BlockVolumeMapper,
  369. outerVolumeSpecName string,
  370. volumeGidValue string,
  371. volumeSpec *volume.Spec) error {
  372. asw.Lock()
  373. defer asw.Unlock()
  374. volumeObj, volumeExists := asw.attachedVolumes[volumeName]
  375. if !volumeExists {
  376. return fmt.Errorf(
  377. "no volume with the name %q exists in the list of attached volumes",
  378. volumeName)
  379. }
  380. podObj, podExists := volumeObj.mountedPods[podName]
  381. if !podExists {
  382. podObj = mountedPod{
  383. podName: podName,
  384. podUID: podUID,
  385. mounter: mounter,
  386. blockVolumeMapper: blockVolumeMapper,
  387. outerVolumeSpecName: outerVolumeSpecName,
  388. volumeGidValue: volumeGidValue,
  389. volumeSpec: volumeSpec,
  390. }
  391. }
  392. // If pod exists, reset remountRequired value
  393. podObj.remountRequired = false
  394. asw.attachedVolumes[volumeName].mountedPods[podName] = podObj
  395. return nil
  396. }
  397. func (asw *actualStateOfWorld) MarkVolumeAsResized(
  398. podName volumetypes.UniquePodName,
  399. volumeName v1.UniqueVolumeName) error {
  400. asw.Lock()
  401. defer asw.Unlock()
  402. volumeObj, volumeExists := asw.attachedVolumes[volumeName]
  403. if !volumeExists {
  404. return fmt.Errorf(
  405. "no volume with the name %q exists in the list of attached volumes",
  406. volumeName)
  407. }
  408. podObj, podExists := volumeObj.mountedPods[podName]
  409. if !podExists {
  410. return fmt.Errorf(
  411. "no pod with the name %q exists in the mounted pods list of volume %s",
  412. podName,
  413. volumeName)
  414. }
  415. klog.V(5).Infof("Volume %s(OuterVolumeSpecName %s) of pod %s has been resized",
  416. volumeName, podObj.outerVolumeSpecName, podName)
  417. podObj.fsResizeRequired = false
  418. asw.attachedVolumes[volumeName].mountedPods[podName] = podObj
  419. return nil
  420. }
  421. func (asw *actualStateOfWorld) MarkRemountRequired(
  422. podName volumetypes.UniquePodName) {
  423. asw.Lock()
  424. defer asw.Unlock()
  425. for volumeName, volumeObj := range asw.attachedVolumes {
  426. for mountedPodName, podObj := range volumeObj.mountedPods {
  427. if mountedPodName != podName {
  428. continue
  429. }
  430. volumePlugin, err :=
  431. asw.volumePluginMgr.FindPluginBySpec(podObj.volumeSpec)
  432. if err != nil || volumePlugin == nil {
  433. // Log and continue processing
  434. klog.Errorf(
  435. "MarkRemountRequired failed to FindPluginBySpec for pod %q (podUid %q) volume: %q (volSpecName: %q)",
  436. podObj.podName,
  437. podObj.podUID,
  438. volumeObj.volumeName,
  439. podObj.volumeSpec.Name())
  440. continue
  441. }
  442. if volumePlugin.RequiresRemount() {
  443. podObj.remountRequired = true
  444. asw.attachedVolumes[volumeName].mountedPods[podName] = podObj
  445. }
  446. }
  447. }
  448. }
  449. func (asw *actualStateOfWorld) MarkFSResizeRequired(
  450. volumeName v1.UniqueVolumeName,
  451. podName volumetypes.UniquePodName) {
  452. asw.Lock()
  453. defer asw.Unlock()
  454. volumeObj, exist := asw.attachedVolumes[volumeName]
  455. if !exist {
  456. klog.Warningf("MarkFSResizeRequired for volume %s failed as volume not exist", volumeName)
  457. return
  458. }
  459. podObj, exist := volumeObj.mountedPods[podName]
  460. if !exist {
  461. klog.Warningf("MarkFSResizeRequired for volume %s failed "+
  462. "as pod(%s) not exist", volumeName, podName)
  463. return
  464. }
  465. volumePlugin, err :=
  466. asw.volumePluginMgr.FindNodeExpandablePluginBySpec(podObj.volumeSpec)
  467. if err != nil || volumePlugin == nil {
  468. // Log and continue processing
  469. klog.Errorf(
  470. "MarkFSResizeRequired failed to find expandable plugin for pod %q volume: %q (volSpecName: %q)",
  471. podObj.podName,
  472. volumeObj.volumeName,
  473. podObj.volumeSpec.Name())
  474. return
  475. }
  476. if volumePlugin.RequiresFSResize() {
  477. if !podObj.fsResizeRequired {
  478. klog.V(3).Infof("PVC volume %s(OuterVolumeSpecName %s) of pod %s requires file system resize",
  479. volumeName, podObj.outerVolumeSpecName, podName)
  480. podObj.fsResizeRequired = true
  481. }
  482. asw.attachedVolumes[volumeName].mountedPods[podName] = podObj
  483. }
  484. }
  485. func (asw *actualStateOfWorld) SetVolumeGloballyMounted(
  486. volumeName v1.UniqueVolumeName, globallyMounted bool, devicePath, deviceMountPath string) error {
  487. asw.Lock()
  488. defer asw.Unlock()
  489. volumeObj, volumeExists := asw.attachedVolumes[volumeName]
  490. if !volumeExists {
  491. return fmt.Errorf(
  492. "no volume with the name %q exists in the list of attached volumes",
  493. volumeName)
  494. }
  495. volumeObj.globallyMounted = globallyMounted
  496. volumeObj.deviceMountPath = deviceMountPath
  497. if devicePath != "" {
  498. volumeObj.devicePath = devicePath
  499. }
  500. asw.attachedVolumes[volumeName] = volumeObj
  501. return nil
  502. }
  503. func (asw *actualStateOfWorld) DeletePodFromVolume(
  504. podName volumetypes.UniquePodName, volumeName v1.UniqueVolumeName) error {
  505. asw.Lock()
  506. defer asw.Unlock()
  507. volumeObj, volumeExists := asw.attachedVolumes[volumeName]
  508. if !volumeExists {
  509. return fmt.Errorf(
  510. "no volume with the name %q exists in the list of attached volumes",
  511. volumeName)
  512. }
  513. _, podExists := volumeObj.mountedPods[podName]
  514. if podExists {
  515. delete(asw.attachedVolumes[volumeName].mountedPods, podName)
  516. }
  517. return nil
  518. }
  519. func (asw *actualStateOfWorld) DeleteVolume(volumeName v1.UniqueVolumeName) error {
  520. asw.Lock()
  521. defer asw.Unlock()
  522. volumeObj, volumeExists := asw.attachedVolumes[volumeName]
  523. if !volumeExists {
  524. return nil
  525. }
  526. if len(volumeObj.mountedPods) != 0 {
  527. return fmt.Errorf(
  528. "failed to DeleteVolume %q, it still has %v mountedPods",
  529. volumeName,
  530. len(volumeObj.mountedPods))
  531. }
  532. delete(asw.attachedVolumes, volumeName)
  533. return nil
  534. }
  535. func (asw *actualStateOfWorld) PodExistsInVolume(
  536. podName volumetypes.UniquePodName,
  537. volumeName v1.UniqueVolumeName) (bool, string, error) {
  538. asw.RLock()
  539. defer asw.RUnlock()
  540. volumeObj, volumeExists := asw.attachedVolumes[volumeName]
  541. if !volumeExists {
  542. return false, "", newVolumeNotAttachedError(volumeName)
  543. }
  544. podObj, podExists := volumeObj.mountedPods[podName]
  545. if podExists {
  546. if podObj.remountRequired {
  547. return true, volumeObj.devicePath, newRemountRequiredError(volumeObj.volumeName, podObj.podName)
  548. }
  549. if podObj.fsResizeRequired &&
  550. utilfeature.DefaultFeatureGate.Enabled(features.ExpandInUsePersistentVolumes) {
  551. return true, volumeObj.devicePath, newFsResizeRequiredError(volumeObj.volumeName, podObj.podName)
  552. }
  553. }
  554. return podExists, volumeObj.devicePath, nil
  555. }
  556. func (asw *actualStateOfWorld) VolumeExistsWithSpecName(podName volumetypes.UniquePodName, volumeSpecName string) bool {
  557. asw.RLock()
  558. defer asw.RUnlock()
  559. for _, volumeObj := range asw.attachedVolumes {
  560. for name, podObj := range volumeObj.mountedPods {
  561. if podName == name && podObj.volumeSpec.Name() == volumeSpecName {
  562. return true
  563. }
  564. }
  565. }
  566. return false
  567. }
  568. func (asw *actualStateOfWorld) VolumeExists(
  569. volumeName v1.UniqueVolumeName) bool {
  570. asw.RLock()
  571. defer asw.RUnlock()
  572. _, volumeExists := asw.attachedVolumes[volumeName]
  573. return volumeExists
  574. }
  575. func (asw *actualStateOfWorld) GetMountedVolumes() []MountedVolume {
  576. asw.RLock()
  577. defer asw.RUnlock()
  578. mountedVolume := make([]MountedVolume, 0 /* len */, len(asw.attachedVolumes) /* cap */)
  579. for _, volumeObj := range asw.attachedVolumes {
  580. for _, podObj := range volumeObj.mountedPods {
  581. mountedVolume = append(
  582. mountedVolume,
  583. getMountedVolume(&podObj, &volumeObj))
  584. }
  585. }
  586. return mountedVolume
  587. }
  588. func (asw *actualStateOfWorld) GetMountedVolumesForPod(
  589. podName volumetypes.UniquePodName) []MountedVolume {
  590. asw.RLock()
  591. defer asw.RUnlock()
  592. mountedVolume := make([]MountedVolume, 0 /* len */, len(asw.attachedVolumes) /* cap */)
  593. for _, volumeObj := range asw.attachedVolumes {
  594. for mountedPodName, podObj := range volumeObj.mountedPods {
  595. if mountedPodName == podName {
  596. mountedVolume = append(
  597. mountedVolume,
  598. getMountedVolume(&podObj, &volumeObj))
  599. }
  600. }
  601. }
  602. return mountedVolume
  603. }
  604. func (asw *actualStateOfWorld) GetGloballyMountedVolumes() []AttachedVolume {
  605. asw.RLock()
  606. defer asw.RUnlock()
  607. globallyMountedVolumes := make(
  608. []AttachedVolume, 0 /* len */, len(asw.attachedVolumes) /* cap */)
  609. for _, volumeObj := range asw.attachedVolumes {
  610. if volumeObj.globallyMounted {
  611. globallyMountedVolumes = append(
  612. globallyMountedVolumes,
  613. asw.newAttachedVolume(&volumeObj))
  614. }
  615. }
  616. return globallyMountedVolumes
  617. }
  618. func (asw *actualStateOfWorld) GetAttachedVolumes() []AttachedVolume {
  619. asw.RLock()
  620. defer asw.RUnlock()
  621. allAttachedVolumes := make(
  622. []AttachedVolume, 0 /* len */, len(asw.attachedVolumes) /* cap */)
  623. for _, volumeObj := range asw.attachedVolumes {
  624. allAttachedVolumes = append(
  625. allAttachedVolumes,
  626. asw.newAttachedVolume(&volumeObj))
  627. }
  628. return allAttachedVolumes
  629. }
  630. func (asw *actualStateOfWorld) GetUnmountedVolumes() []AttachedVolume {
  631. asw.RLock()
  632. defer asw.RUnlock()
  633. unmountedVolumes := make([]AttachedVolume, 0 /* len */, len(asw.attachedVolumes) /* cap */)
  634. for _, volumeObj := range asw.attachedVolumes {
  635. if len(volumeObj.mountedPods) == 0 {
  636. unmountedVolumes = append(
  637. unmountedVolumes,
  638. asw.newAttachedVolume(&volumeObj))
  639. }
  640. }
  641. return unmountedVolumes
  642. }
  643. func (asw *actualStateOfWorld) GetPods() map[volumetypes.UniquePodName]bool {
  644. asw.RLock()
  645. defer asw.RUnlock()
  646. podList := make(map[volumetypes.UniquePodName]bool)
  647. for _, volumeObj := range asw.attachedVolumes {
  648. for podName := range volumeObj.mountedPods {
  649. if !podList[podName] {
  650. podList[podName] = true
  651. }
  652. }
  653. }
  654. return podList
  655. }
  656. func (asw *actualStateOfWorld) newAttachedVolume(
  657. attachedVolume *attachedVolume) AttachedVolume {
  658. return AttachedVolume{
  659. AttachedVolume: operationexecutor.AttachedVolume{
  660. VolumeName: attachedVolume.volumeName,
  661. VolumeSpec: attachedVolume.spec,
  662. NodeName: asw.nodeName,
  663. PluginIsAttachable: attachedVolume.pluginIsAttachable,
  664. DevicePath: attachedVolume.devicePath,
  665. DeviceMountPath: attachedVolume.deviceMountPath,
  666. PluginName: attachedVolume.pluginName},
  667. GloballyMounted: attachedVolume.globallyMounted,
  668. }
  669. }
  670. // Compile-time check to ensure volumeNotAttachedError implements the error interface
  671. var _ error = volumeNotAttachedError{}
  672. // volumeNotAttachedError is an error returned when PodExistsInVolume() fails to
  673. // find specified volume in the list of attached volumes.
  674. type volumeNotAttachedError struct {
  675. volumeName v1.UniqueVolumeName
  676. }
  677. func (err volumeNotAttachedError) Error() string {
  678. return fmt.Sprintf(
  679. "volumeName %q does not exist in the list of attached volumes",
  680. err.volumeName)
  681. }
  682. func newVolumeNotAttachedError(volumeName v1.UniqueVolumeName) error {
  683. return volumeNotAttachedError{
  684. volumeName: volumeName,
  685. }
  686. }
  687. // Compile-time check to ensure remountRequiredError implements the error interface
  688. var _ error = remountRequiredError{}
  689. // remountRequiredError is an error returned when PodExistsInVolume() found
  690. // volume/pod attached/mounted but remountRequired was true, indicating the
  691. // given volume should be remounted to the pod to reflect changes in the
  692. // referencing pod.
  693. type remountRequiredError struct {
  694. volumeName v1.UniqueVolumeName
  695. podName volumetypes.UniquePodName
  696. }
  697. func (err remountRequiredError) Error() string {
  698. return fmt.Sprintf(
  699. "volumeName %q is mounted to %q but should be remounted",
  700. err.volumeName, err.podName)
  701. }
  702. func newRemountRequiredError(
  703. volumeName v1.UniqueVolumeName, podName volumetypes.UniquePodName) error {
  704. return remountRequiredError{
  705. volumeName: volumeName,
  706. podName: podName,
  707. }
  708. }
  709. // fsResizeRequiredError is an error returned when PodExistsInVolume() found
  710. // volume/pod attached/mounted but fsResizeRequired was true, indicating the
  711. // given volume receives an resize request after attached/mounted.
  712. type fsResizeRequiredError struct {
  713. volumeName v1.UniqueVolumeName
  714. podName volumetypes.UniquePodName
  715. }
  716. func (err fsResizeRequiredError) Error() string {
  717. return fmt.Sprintf(
  718. "volumeName %q mounted to %q needs to resize file system",
  719. err.volumeName, err.podName)
  720. }
  721. func newFsResizeRequiredError(
  722. volumeName v1.UniqueVolumeName, podName volumetypes.UniquePodName) error {
  723. return fsResizeRequiredError{
  724. volumeName: volumeName,
  725. podName: podName,
  726. }
  727. }
  728. // IsFSResizeRequiredError returns true if the specified error is a
  729. // fsResizeRequiredError.
  730. func IsFSResizeRequiredError(err error) bool {
  731. _, ok := err.(fsResizeRequiredError)
  732. return ok
  733. }
  734. // getMountedVolume constructs and returns a MountedVolume object from the given
  735. // mountedPod and attachedVolume objects.
  736. func getMountedVolume(
  737. mountedPod *mountedPod, attachedVolume *attachedVolume) MountedVolume {
  738. return MountedVolume{
  739. MountedVolume: operationexecutor.MountedVolume{
  740. PodName: mountedPod.podName,
  741. VolumeName: attachedVolume.volumeName,
  742. InnerVolumeSpecName: mountedPod.volumeSpec.Name(),
  743. OuterVolumeSpecName: mountedPod.outerVolumeSpecName,
  744. PluginName: attachedVolume.pluginName,
  745. PodUID: mountedPod.podUID,
  746. Mounter: mountedPod.mounter,
  747. BlockVolumeMapper: mountedPod.blockVolumeMapper,
  748. VolumeGidValue: mountedPod.volumeGidValue,
  749. VolumeSpec: mountedPod.volumeSpec,
  750. DeviceMountPath: attachedVolume.deviceMountPath}}
  751. }