glusterfs.go 42 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236
  1. /*
  2. Copyright 2015 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. package glusterfs
  14. import (
  15. "context"
  16. "fmt"
  17. "math"
  18. "math/rand"
  19. "net"
  20. "os"
  21. "path/filepath"
  22. "runtime"
  23. "strconv"
  24. dstrings "strings"
  25. "sync"
  26. gcli "github.com/heketi/heketi/client/api/go-client"
  27. gapi "github.com/heketi/heketi/pkg/glusterfs/api"
  28. "k8s.io/klog"
  29. "k8s.io/utils/mount"
  30. utilstrings "k8s.io/utils/strings"
  31. v1 "k8s.io/api/core/v1"
  32. "k8s.io/apimachinery/pkg/api/errors"
  33. "k8s.io/apimachinery/pkg/api/resource"
  34. metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
  35. "k8s.io/apimachinery/pkg/labels"
  36. "k8s.io/apimachinery/pkg/types"
  37. "k8s.io/apimachinery/pkg/util/sets"
  38. "k8s.io/apimachinery/pkg/util/uuid"
  39. clientset "k8s.io/client-go/kubernetes"
  40. volumehelpers "k8s.io/cloud-provider/volume/helpers"
  41. v1helper "k8s.io/kubernetes/pkg/apis/core/v1/helper"
  42. "k8s.io/kubernetes/pkg/volume"
  43. volutil "k8s.io/kubernetes/pkg/volume/util"
  44. )
  45. // ProbeVolumePlugins is the primary entrypoint for volume plugins.
  46. func ProbeVolumePlugins() []volume.VolumePlugin {
  47. return []volume.VolumePlugin{&glusterfsPlugin{host: nil, gidTable: make(map[string]*MinMaxAllocator)}}
  48. }
  49. type glusterfsPlugin struct {
  50. host volume.VolumeHost
  51. gidTable map[string]*MinMaxAllocator
  52. gidTableLock sync.Mutex
  53. }
  54. var _ volume.VolumePlugin = &glusterfsPlugin{}
  55. var _ volume.PersistentVolumePlugin = &glusterfsPlugin{}
  56. var _ volume.DeletableVolumePlugin = &glusterfsPlugin{}
  57. var _ volume.ProvisionableVolumePlugin = &glusterfsPlugin{}
  58. var _ volume.ExpandableVolumePlugin = &glusterfsPlugin{}
  59. var _ volume.Provisioner = &glusterfsVolumeProvisioner{}
  60. var _ volume.Deleter = &glusterfsVolumeDeleter{}
  61. const (
  62. glusterfsPluginName = "kubernetes.io/glusterfs"
  63. volPrefix = "vol_"
  64. dynamicEpSvcPrefix = "glusterfs-dynamic"
  65. replicaCount = 3
  66. durabilityType = "replicate"
  67. secretKeyName = "key" // key name used in secret
  68. gciLinuxGlusterMountBinaryPath = "/sbin/mount.glusterfs"
  69. defaultGidMin = 2000
  70. defaultGidMax = math.MaxInt32
  71. // maxCustomEpNamePrefix is the maximum number of chars.
  72. // which can be used as ep/svc name prefix. This number is carved
  73. // out from below formula.
  74. // max length of name of an ep - length of pvc uuid
  75. // where max length of name of an ep is 63 and length of uuid is 37
  76. maxCustomEpNamePrefixLen = 26
  77. // absoluteGidMin/Max are currently the same as the
  78. // default values, but they play a different role and
  79. // could take a different value. Only thing we need is:
  80. // absGidMin <= defGidMin <= defGidMax <= absGidMax
  81. absoluteGidMin = 2000
  82. absoluteGidMax = math.MaxInt32
  83. heketiAnn = "heketi-dynamic-provisioner"
  84. glusterTypeAnn = "gluster.org/type"
  85. glusterDescAnn = "Gluster-Internal: Dynamically provisioned PV"
  86. heketiVolIDAnn = "gluster.kubernetes.io/heketi-volume-id"
  87. // Error string returned by heketi
  88. errIDNotFound = "Id not found"
  89. )
  90. func (plugin *glusterfsPlugin) Init(host volume.VolumeHost) error {
  91. plugin.host = host
  92. return nil
  93. }
  94. func (plugin *glusterfsPlugin) GetPluginName() string {
  95. return glusterfsPluginName
  96. }
  97. func (plugin *glusterfsPlugin) GetVolumeName(spec *volume.Spec) (string, error) {
  98. return "", fmt.Errorf("GetVolumeName() is unimplemented for GlusterFS")
  99. }
  100. func (plugin *glusterfsPlugin) CanSupport(spec *volume.Spec) bool {
  101. return (spec.PersistentVolume != nil && spec.PersistentVolume.Spec.Glusterfs != nil) ||
  102. (spec.Volume != nil && spec.Volume.Glusterfs != nil)
  103. }
  104. func (plugin *glusterfsPlugin) RequiresRemount() bool {
  105. return false
  106. }
  107. func (plugin *glusterfsPlugin) SupportsMountOption() bool {
  108. return true
  109. }
  110. func (plugin *glusterfsPlugin) SupportsBulkVolumeVerification() bool {
  111. return false
  112. }
  113. func (plugin *glusterfsPlugin) RequiresFSResize() bool {
  114. return false
  115. }
  116. func (plugin *glusterfsPlugin) GetAccessModes() []v1.PersistentVolumeAccessMode {
  117. return []v1.PersistentVolumeAccessMode{
  118. v1.ReadWriteOnce,
  119. v1.ReadOnlyMany,
  120. v1.ReadWriteMany,
  121. }
  122. }
  123. func (plugin *glusterfsPlugin) NewMounter(spec *volume.Spec, pod *v1.Pod, _ volume.VolumeOptions) (volume.Mounter, error) {
  124. epName, epNamespace, err := plugin.getEndpointNameAndNamespace(spec, pod.Namespace)
  125. if err != nil {
  126. return nil, err
  127. }
  128. kubeClient := plugin.host.GetKubeClient()
  129. if kubeClient == nil {
  130. return nil, fmt.Errorf("failed to get kube client to initialize mounter")
  131. }
  132. ep, err := kubeClient.CoreV1().Endpoints(epNamespace).Get(context.TODO(), epName, metav1.GetOptions{})
  133. if err != nil {
  134. klog.Errorf("failed to get endpoint %s: %v", epName, err)
  135. return nil, err
  136. }
  137. klog.V(4).Infof("glusterfs pv endpoint %v", ep)
  138. return plugin.newMounterInternal(spec, ep, pod, plugin.host.GetMounter(plugin.GetPluginName()))
  139. }
  140. func (plugin *glusterfsPlugin) getEndpointNameAndNamespace(spec *volume.Spec, defaultNamespace string) (string, string, error) {
  141. if spec.Volume != nil && spec.Volume.Glusterfs != nil {
  142. endpoints := spec.Volume.Glusterfs.EndpointsName
  143. if endpoints == "" {
  144. return "", "", fmt.Errorf("no glusterFS endpoint specified")
  145. }
  146. return endpoints, defaultNamespace, nil
  147. } else if spec.PersistentVolume != nil &&
  148. spec.PersistentVolume.Spec.Glusterfs != nil {
  149. endpoints := spec.PersistentVolume.Spec.Glusterfs.EndpointsName
  150. endpointsNs := defaultNamespace
  151. overriddenNs := spec.PersistentVolume.Spec.Glusterfs.EndpointsNamespace
  152. if overriddenNs != nil {
  153. if len(*overriddenNs) > 0 {
  154. endpointsNs = *overriddenNs
  155. } else {
  156. return "", "", fmt.Errorf("endpointnamespace field set, but no endpointnamespace specified")
  157. }
  158. }
  159. return endpoints, endpointsNs, nil
  160. }
  161. return "", "", fmt.Errorf("spec does not reference a GlusterFS volume type")
  162. }
  163. func (plugin *glusterfsPlugin) newMounterInternal(spec *volume.Spec, ep *v1.Endpoints, pod *v1.Pod, mounter mount.Interface) (volume.Mounter, error) {
  164. volPath, readOnly, err := getVolumeInfo(spec)
  165. if err != nil {
  166. klog.Errorf("failed to get volumesource: %v", err)
  167. return nil, err
  168. }
  169. return &glusterfsMounter{
  170. glusterfs: &glusterfs{
  171. volName: spec.Name(),
  172. mounter: mounter,
  173. pod: pod,
  174. plugin: plugin,
  175. MetricsProvider: volume.NewMetricsStatFS(plugin.host.GetPodVolumeDir(pod.UID, utilstrings.EscapeQualifiedName(glusterfsPluginName), spec.Name())),
  176. },
  177. hosts: ep,
  178. path: volPath,
  179. readOnly: readOnly,
  180. mountOptions: volutil.MountOptionFromSpec(spec),
  181. }, nil
  182. }
  183. func (plugin *glusterfsPlugin) NewUnmounter(volName string, podUID types.UID) (volume.Unmounter, error) {
  184. return plugin.newUnmounterInternal(volName, podUID, plugin.host.GetMounter(plugin.GetPluginName()))
  185. }
  186. func (plugin *glusterfsPlugin) newUnmounterInternal(volName string, podUID types.UID, mounter mount.Interface) (volume.Unmounter, error) {
  187. return &glusterfsUnmounter{&glusterfs{
  188. volName: volName,
  189. mounter: mounter,
  190. pod: &v1.Pod{ObjectMeta: metav1.ObjectMeta{UID: podUID}},
  191. plugin: plugin,
  192. MetricsProvider: volume.NewMetricsStatFS(plugin.host.GetPodVolumeDir(podUID, utilstrings.EscapeQualifiedName(glusterfsPluginName), volName)),
  193. }}, nil
  194. }
  195. func (plugin *glusterfsPlugin) ConstructVolumeSpec(volumeName, mountPath string) (*volume.Spec, error) {
  196. // To reconstruct volume spec we need endpoint where fetching endpoint from mount
  197. // string looks to be impossible, so returning error.
  198. return nil, fmt.Errorf("impossible to reconstruct glusterfs volume spec from volume mountpath")
  199. }
  200. // Glusterfs volumes represent a bare host file or directory mount of an Glusterfs export.
  201. type glusterfs struct {
  202. volName string
  203. pod *v1.Pod
  204. mounter mount.Interface
  205. plugin *glusterfsPlugin
  206. volume.MetricsProvider
  207. }
  208. type glusterfsMounter struct {
  209. *glusterfs
  210. hosts *v1.Endpoints
  211. path string
  212. readOnly bool
  213. mountOptions []string
  214. }
  215. var _ volume.Mounter = &glusterfsMounter{}
  216. func (b *glusterfsMounter) GetAttributes() volume.Attributes {
  217. return volume.Attributes{
  218. ReadOnly: b.readOnly,
  219. Managed: false,
  220. SupportsSELinux: false,
  221. }
  222. }
  223. // Checks prior to mount operations to verify that the required components (binaries, etc.)
  224. // to mount the volume are available on the underlying node.
  225. // If not, it returns an error
  226. func (b *glusterfsMounter) CanMount() error {
  227. exe := b.plugin.host.GetExec(b.plugin.GetPluginName())
  228. switch runtime.GOOS {
  229. case "linux":
  230. if _, err := exe.Command("test", "-x", gciLinuxGlusterMountBinaryPath).CombinedOutput(); err != nil {
  231. return fmt.Errorf("required binary %s is missing", gciLinuxGlusterMountBinaryPath)
  232. }
  233. }
  234. return nil
  235. }
  236. // SetUp attaches the disk and bind mounts to the volume path.
  237. func (b *glusterfsMounter) SetUp(mounterArgs volume.MounterArgs) error {
  238. return b.SetUpAt(b.GetPath(), mounterArgs)
  239. }
  240. func (b *glusterfsMounter) SetUpAt(dir string, mounterArgs volume.MounterArgs) error {
  241. notMnt, err := b.mounter.IsLikelyNotMountPoint(dir)
  242. klog.V(4).Infof("mount setup: %s %v %v", dir, !notMnt, err)
  243. if err != nil && !os.IsNotExist(err) {
  244. return err
  245. }
  246. if !notMnt {
  247. return nil
  248. }
  249. if err := os.MkdirAll(dir, 0750); err != nil {
  250. return err
  251. }
  252. err = b.setUpAtInternal(dir)
  253. if err == nil {
  254. return nil
  255. }
  256. // Cleanup upon failure.
  257. mount.CleanupMountPoint(dir, b.mounter, false)
  258. return err
  259. }
  260. func (glusterfsVolume *glusterfs) GetPath() string {
  261. name := glusterfsPluginName
  262. return glusterfsVolume.plugin.host.GetPodVolumeDir(glusterfsVolume.pod.UID, utilstrings.EscapeQualifiedName(name), glusterfsVolume.volName)
  263. }
  264. type glusterfsUnmounter struct {
  265. *glusterfs
  266. }
  267. var _ volume.Unmounter = &glusterfsUnmounter{}
  268. func (c *glusterfsUnmounter) TearDown() error {
  269. return c.TearDownAt(c.GetPath())
  270. }
  271. func (c *glusterfsUnmounter) TearDownAt(dir string) error {
  272. return mount.CleanupMountPoint(dir, c.mounter, false)
  273. }
  274. func (b *glusterfsMounter) setUpAtInternal(dir string) error {
  275. var errs error
  276. options := []string{}
  277. hasLogFile := false
  278. hasLogLevel := false
  279. log := ""
  280. if b.readOnly {
  281. options = append(options, "ro")
  282. }
  283. // Check for log-file,log-level options existence in user supplied mount options, if provided, use those.
  284. for _, userOpt := range b.mountOptions {
  285. switch {
  286. case dstrings.HasPrefix(userOpt, "log-file"):
  287. klog.V(4).Infof("log-file mount option has provided")
  288. hasLogFile = true
  289. case dstrings.HasPrefix(userOpt, "log-level"):
  290. klog.V(4).Infof("log-level mount option has provided")
  291. hasLogLevel = true
  292. }
  293. }
  294. // If logfile has not been provided, create driver specific log file.
  295. if !hasLogFile {
  296. p := filepath.Join(b.glusterfs.plugin.host.GetPluginDir(glusterfsPluginName), b.glusterfs.volName)
  297. if err := os.MkdirAll(p, 0750); err != nil {
  298. return fmt.Errorf("failed to create directory %v: %v", p, err)
  299. }
  300. // adding log-level ERROR to remove noise
  301. // and more specific log path so each pod has
  302. // its own log based on PV + Pod
  303. log = filepath.Join(p, b.pod.Name+"-glusterfs.log")
  304. // Use derived log file in gluster fuse mount
  305. options = append(options, "log-file="+log)
  306. }
  307. if !hasLogLevel {
  308. options = append(options, "log-level=ERROR")
  309. }
  310. var addrlist []string
  311. if b.hosts == nil {
  312. return fmt.Errorf("glusterfs endpoint is nil in mounter")
  313. }
  314. addr := sets.String{}
  315. if b.hosts.Subsets != nil {
  316. for _, s := range b.hosts.Subsets {
  317. for _, a := range s.Addresses {
  318. if !addr.Has(a.IP) {
  319. addr.Insert(a.IP)
  320. addrlist = append(addrlist, a.IP)
  321. }
  322. }
  323. }
  324. }
  325. if (len(addrlist) > 0) && (addrlist[0] != "") {
  326. ip := addrlist[rand.Intn(len(addrlist))]
  327. // Add backup-volfile-servers and auto_unmount options.
  328. // When ip is also in backup-volfile-servers, there will be a warning:
  329. // "gf_remember_backup_volfile_server] 0-glusterfs: failed to set volfile server: File exists".
  330. addr.Delete(ip)
  331. backups := addr.List()
  332. // Avoid an invalid empty backup-volfile-servers option.
  333. if len(backups) > 0 {
  334. options = append(options, "backup-volfile-servers="+dstrings.Join(addrlist[:], ":"))
  335. }
  336. options = append(options, "auto_unmount")
  337. mountOptions := volutil.JoinMountOptions(b.mountOptions, options)
  338. // with `backup-volfile-servers` mount option in place, it is not required to
  339. // iterate over all the servers in the addrlist. A mount attempt with this option
  340. // will fetch all the servers mentioned in the backup-volfile-servers list.
  341. // Refer to backup-volfile-servers @ http://docs.gluster.org/en/latest/Administrator%20Guide/Setting%20Up%20Clients/
  342. errs = b.mounter.Mount(ip+":"+b.path, dir, "glusterfs", mountOptions)
  343. if errs == nil {
  344. klog.Infof("successfully mounted directory %s", dir)
  345. return nil
  346. }
  347. if dstrings.Contains(errs.Error(), "Invalid option auto_unmount") ||
  348. dstrings.Contains(errs.Error(), "Invalid argument") {
  349. // Give a try without `auto_unmount` mount option, because
  350. // it could be that gluster fuse client is older version and
  351. // mount.glusterfs is unaware of `auto_unmount`.
  352. noAutoMountOptions := make([]string, 0, len(mountOptions))
  353. for _, opt := range mountOptions {
  354. if opt != "auto_unmount" {
  355. noAutoMountOptions = append(noAutoMountOptions, opt)
  356. }
  357. }
  358. errs = b.mounter.Mount(ip+":"+b.path, dir, "glusterfs", noAutoMountOptions)
  359. if errs == nil {
  360. klog.Infof("successfully mounted %s", dir)
  361. return nil
  362. }
  363. }
  364. } else {
  365. return fmt.Errorf("failed to execute mount command:[no valid ipaddress found in endpoint address list]")
  366. }
  367. // Failed mount scenario.
  368. // Since glusterfs does not return error text
  369. // it all goes in a log file, we will read the log file
  370. logErr := readGlusterLog(log, b.pod.Name)
  371. if logErr != nil {
  372. return fmt.Errorf("mount failed: %v, the following error information was pulled from the glusterfs log to help diagnose this issue: %v", errs, logErr)
  373. }
  374. return fmt.Errorf("mount failed: %v", errs)
  375. }
  376. //getVolumeInfo returns 'path' and 'readonly' field values from the provided glusterfs spec.
  377. func getVolumeInfo(spec *volume.Spec) (string, bool, error) {
  378. if spec.Volume != nil && spec.Volume.Glusterfs != nil {
  379. return spec.Volume.Glusterfs.Path, spec.Volume.Glusterfs.ReadOnly, nil
  380. } else if spec.PersistentVolume != nil &&
  381. spec.PersistentVolume.Spec.Glusterfs != nil {
  382. return spec.PersistentVolume.Spec.Glusterfs.Path, spec.ReadOnly, nil
  383. }
  384. return "", false, fmt.Errorf("spec does not reference a Glusterfs volume type")
  385. }
  386. func (plugin *glusterfsPlugin) NewProvisioner(options volume.VolumeOptions) (volume.Provisioner, error) {
  387. return plugin.newProvisionerInternal(options)
  388. }
  389. func (plugin *glusterfsPlugin) newProvisionerInternal(options volume.VolumeOptions) (volume.Provisioner, error) {
  390. return &glusterfsVolumeProvisioner{
  391. glusterfsMounter: &glusterfsMounter{
  392. glusterfs: &glusterfs{
  393. plugin: plugin,
  394. },
  395. },
  396. options: options,
  397. }, nil
  398. }
  399. type provisionerConfig struct {
  400. url string
  401. user string
  402. userKey string
  403. secretNamespace string
  404. secretName string
  405. secretValue string
  406. clusterID string
  407. gidMin int
  408. gidMax int
  409. volumeType gapi.VolumeDurabilityInfo
  410. volumeOptions []string
  411. volumeNamePrefix string
  412. thinPoolSnapFactor float32
  413. customEpNamePrefix string
  414. }
  415. type glusterfsVolumeProvisioner struct {
  416. *glusterfsMounter
  417. provisionerConfig
  418. options volume.VolumeOptions
  419. }
  420. func convertGid(gidString string) (int, error) {
  421. gid64, err := strconv.ParseInt(gidString, 10, 32)
  422. if err != nil {
  423. return 0, fmt.Errorf("failed to parse gid %v: %v", gidString, err)
  424. }
  425. if gid64 < 0 {
  426. return 0, fmt.Errorf("negative GIDs %v are not allowed", gidString)
  427. }
  428. // ParseInt returns a int64, but since we parsed only
  429. // for 32 bit, we can cast to int without loss:
  430. gid := int(gid64)
  431. return gid, nil
  432. }
  433. func convertVolumeParam(volumeString string) (int, error) {
  434. count, err := strconv.Atoi(volumeString)
  435. if err != nil {
  436. return 0, fmt.Errorf("failed to parse volumestring %q: %v", volumeString, err)
  437. }
  438. if count < 0 {
  439. return 0, fmt.Errorf("negative values are not allowed")
  440. }
  441. return count, nil
  442. }
  443. func (plugin *glusterfsPlugin) NewDeleter(spec *volume.Spec) (volume.Deleter, error) {
  444. return plugin.newDeleterInternal(spec)
  445. }
  446. func (plugin *glusterfsPlugin) newDeleterInternal(spec *volume.Spec) (volume.Deleter, error) {
  447. if spec.PersistentVolume != nil && spec.PersistentVolume.Spec.Glusterfs == nil {
  448. return nil, fmt.Errorf("spec.PersistentVolume.Spec.Glusterfs is nil")
  449. }
  450. return &glusterfsVolumeDeleter{
  451. glusterfsMounter: &glusterfsMounter{
  452. glusterfs: &glusterfs{
  453. volName: spec.Name(),
  454. plugin: plugin,
  455. },
  456. path: spec.PersistentVolume.Spec.Glusterfs.Path,
  457. },
  458. spec: spec.PersistentVolume,
  459. }, nil
  460. }
  461. type glusterfsVolumeDeleter struct {
  462. *glusterfsMounter
  463. provisionerConfig
  464. spec *v1.PersistentVolume
  465. }
  466. func (d *glusterfsVolumeDeleter) GetPath() string {
  467. name := glusterfsPluginName
  468. return d.plugin.host.GetPodVolumeDir(d.glusterfsMounter.glusterfs.pod.UID, utilstrings.EscapeQualifiedName(name), d.glusterfsMounter.glusterfs.volName)
  469. }
  470. // Traverse the PVs, fetching all the GIDs from those
  471. // in a given storage class, and mark them in the table.
  472. func (plugin *glusterfsPlugin) collectGids(className string, gidTable *MinMaxAllocator) error {
  473. kubeClient := plugin.host.GetKubeClient()
  474. if kubeClient == nil {
  475. return fmt.Errorf("failed to get kube client when collecting gids")
  476. }
  477. pvList, err := kubeClient.CoreV1().PersistentVolumes().List(context.TODO(), metav1.ListOptions{LabelSelector: labels.Everything().String()})
  478. if err != nil {
  479. return fmt.Errorf("failed to get existing persistent volumes")
  480. }
  481. for _, pv := range pvList.Items {
  482. if v1helper.GetPersistentVolumeClass(&pv) != className {
  483. continue
  484. }
  485. pvName := pv.ObjectMeta.Name
  486. gidStr, ok := pv.Annotations[volutil.VolumeGidAnnotationKey]
  487. if !ok {
  488. klog.Warningf("no GID found in pv %v", pvName)
  489. continue
  490. }
  491. gid, err := convertGid(gidStr)
  492. if err != nil {
  493. klog.Errorf("failed to parse gid %s: %v", gidStr, err)
  494. continue
  495. }
  496. _, err = gidTable.Allocate(gid)
  497. if err == ErrConflict {
  498. klog.Warningf("GID %v found in pv %v was already allocated", gid, pvName)
  499. } else if err != nil {
  500. return fmt.Errorf("failed to store gid %v found in pv %v: %v", gid, pvName, err)
  501. }
  502. }
  503. return nil
  504. }
  505. // Return the gid table for a storage class.
  506. // - If this is the first time, fill it with all the gids
  507. // used in PVs of this storage class by traversing the PVs.
  508. // - Adapt the range of the table to the current range of the SC.
  509. func (plugin *glusterfsPlugin) getGidTable(className string, min int, max int) (*MinMaxAllocator, error) {
  510. plugin.gidTableLock.Lock()
  511. gidTable, ok := plugin.gidTable[className]
  512. plugin.gidTableLock.Unlock()
  513. if ok {
  514. err := gidTable.SetRange(min, max)
  515. if err != nil {
  516. return nil, err
  517. }
  518. return gidTable, nil
  519. }
  520. // create a new table and fill it
  521. newGidTable, err := NewMinMaxAllocator(0, absoluteGidMax)
  522. if err != nil {
  523. return nil, err
  524. }
  525. // collect gids with the full range
  526. err = plugin.collectGids(className, newGidTable)
  527. if err != nil {
  528. return nil, err
  529. }
  530. // and only reduce the range afterwards
  531. err = newGidTable.SetRange(min, max)
  532. if err != nil {
  533. return nil, err
  534. }
  535. // if in the meantime a table appeared, use it
  536. plugin.gidTableLock.Lock()
  537. defer plugin.gidTableLock.Unlock()
  538. gidTable, ok = plugin.gidTable[className]
  539. if ok {
  540. err = gidTable.SetRange(min, max)
  541. if err != nil {
  542. return nil, err
  543. }
  544. return gidTable, nil
  545. }
  546. plugin.gidTable[className] = newGidTable
  547. return newGidTable, nil
  548. }
  549. func (d *glusterfsVolumeDeleter) getGid() (int, bool, error) {
  550. gidStr, ok := d.spec.Annotations[volutil.VolumeGidAnnotationKey]
  551. if !ok {
  552. return 0, false, nil
  553. }
  554. gid, err := convertGid(gidStr)
  555. return gid, true, err
  556. }
  557. func (d *glusterfsVolumeDeleter) Delete() error {
  558. klog.V(2).Infof("delete volume %s", d.glusterfsMounter.path)
  559. volumeName := d.glusterfsMounter.path
  560. volumeID, err := getVolumeID(d.spec, volumeName)
  561. if err != nil {
  562. return fmt.Errorf("failed to get volumeID: %v", err)
  563. }
  564. class, err := volutil.GetClassForVolume(d.plugin.host.GetKubeClient(), d.spec)
  565. if err != nil {
  566. return err
  567. }
  568. cfg, err := parseClassParameters(class.Parameters, d.plugin.host.GetKubeClient())
  569. if err != nil {
  570. return err
  571. }
  572. d.provisionerConfig = *cfg
  573. klog.V(4).Infof("deleting volume %q", volumeID)
  574. gid, exists, err := d.getGid()
  575. if err != nil {
  576. klog.Error(err)
  577. } else if exists {
  578. gidTable, err := d.plugin.getGidTable(class.Name, cfg.gidMin, cfg.gidMax)
  579. if err != nil {
  580. return fmt.Errorf("failed to get gidTable: %v", err)
  581. }
  582. err = gidTable.Release(gid)
  583. if err != nil {
  584. return fmt.Errorf("failed to release gid %v: %v", gid, err)
  585. }
  586. }
  587. cli := gcli.NewClient(d.url, d.user, d.secretValue)
  588. if cli == nil {
  589. klog.Errorf("failed to create glusterfs REST client")
  590. return fmt.Errorf("failed to create glusterfs REST client, REST server authentication failed")
  591. }
  592. err = cli.VolumeDelete(volumeID)
  593. if err != nil {
  594. if dstrings.TrimSpace(err.Error()) != errIDNotFound {
  595. klog.Errorf("failed to delete volume %s: %v", volumeName, err)
  596. return fmt.Errorf("failed to delete volume %s: %v", volumeName, err)
  597. }
  598. klog.V(2).Infof("volume %s not present in heketi, ignoring", volumeName)
  599. }
  600. klog.V(2).Infof("volume %s deleted successfully", volumeName)
  601. //Deleter takes endpoint and namespace from pv spec.
  602. pvSpec := d.spec.Spec
  603. var dynamicEndpoint, dynamicNamespace string
  604. if pvSpec.ClaimRef == nil {
  605. klog.Errorf("ClaimRef is nil")
  606. return fmt.Errorf("ClaimRef is nil")
  607. }
  608. if pvSpec.ClaimRef.Namespace == "" {
  609. klog.Errorf("namespace is nil")
  610. return fmt.Errorf("namespace is nil")
  611. }
  612. dynamicNamespace = pvSpec.ClaimRef.Namespace
  613. if pvSpec.Glusterfs.EndpointsName != "" {
  614. dynamicEndpoint = pvSpec.Glusterfs.EndpointsName
  615. }
  616. klog.V(3).Infof("dynamic namespace and endpoint %v/%v", dynamicNamespace, dynamicEndpoint)
  617. err = d.deleteEndpointService(dynamicNamespace, dynamicEndpoint)
  618. if err != nil {
  619. klog.Errorf("failed to delete endpoint/service %v/%v: %v", dynamicNamespace, dynamicEndpoint, err)
  620. } else {
  621. klog.V(1).Infof("endpoint %v/%v is deleted successfully ", dynamicNamespace, dynamicEndpoint)
  622. }
  623. return nil
  624. }
  625. func (p *glusterfsVolumeProvisioner) Provision(selectedNode *v1.Node, allowedTopologies []v1.TopologySelectorTerm) (*v1.PersistentVolume, error) {
  626. if !volutil.AccessModesContainedInAll(p.plugin.GetAccessModes(), p.options.PVC.Spec.AccessModes) {
  627. return nil, fmt.Errorf("invalid AccessModes %v: only AccessModes %v are supported", p.options.PVC.Spec.AccessModes, p.plugin.GetAccessModes())
  628. }
  629. if p.options.PVC.Spec.Selector != nil {
  630. klog.V(4).Infof("not able to parse your claim Selector")
  631. return nil, fmt.Errorf("not able to parse your claim Selector")
  632. }
  633. if volutil.CheckPersistentVolumeClaimModeBlock(p.options.PVC) {
  634. return nil, fmt.Errorf("%s does not support block volume provisioning", p.plugin.GetPluginName())
  635. }
  636. klog.V(4).Infof("provision volume with options %v", p.options)
  637. scName := v1helper.GetPersistentVolumeClaimClass(p.options.PVC)
  638. cfg, err := parseClassParameters(p.options.Parameters, p.plugin.host.GetKubeClient())
  639. if err != nil {
  640. return nil, err
  641. }
  642. p.provisionerConfig = *cfg
  643. gidTable, err := p.plugin.getGidTable(scName, cfg.gidMin, cfg.gidMax)
  644. if err != nil {
  645. return nil, fmt.Errorf("failed to get gidTable: %v", err)
  646. }
  647. gid, _, err := gidTable.AllocateNext()
  648. if err != nil {
  649. klog.Errorf("failed to reserve GID from table: %v", err)
  650. return nil, fmt.Errorf("failed to reserve GID from table: %v", err)
  651. }
  652. klog.V(2).Infof("allocated GID %d for PVC %s", gid, p.options.PVC.Name)
  653. glusterfs, sizeGiB, volID, err := p.CreateVolume(gid)
  654. if err != nil {
  655. if releaseErr := gidTable.Release(gid); releaseErr != nil {
  656. klog.Errorf("error when releasing GID in storageclass %s: %v", scName, releaseErr)
  657. }
  658. return nil, fmt.Errorf("failed to create volume: %v", err)
  659. }
  660. mode := v1.PersistentVolumeFilesystem
  661. pv := new(v1.PersistentVolume)
  662. pv.Spec.PersistentVolumeSource.Glusterfs = glusterfs
  663. pv.Spec.PersistentVolumeReclaimPolicy = p.options.PersistentVolumeReclaimPolicy
  664. pv.Spec.AccessModes = p.options.PVC.Spec.AccessModes
  665. pv.Spec.VolumeMode = &mode
  666. if len(pv.Spec.AccessModes) == 0 {
  667. pv.Spec.AccessModes = p.plugin.GetAccessModes()
  668. }
  669. pv.Spec.MountOptions = p.options.MountOptions
  670. gidStr := strconv.FormatInt(int64(gid), 10)
  671. pv.Annotations = map[string]string{
  672. volutil.VolumeGidAnnotationKey: gidStr,
  673. volutil.VolumeDynamicallyCreatedByKey: heketiAnn,
  674. glusterTypeAnn: "file",
  675. "Description": glusterDescAnn,
  676. heketiVolIDAnn: volID,
  677. }
  678. pv.Spec.Capacity = v1.ResourceList{
  679. v1.ResourceName(v1.ResourceStorage): resource.MustParse(fmt.Sprintf("%dGi", sizeGiB)),
  680. }
  681. return pv, nil
  682. }
  683. func (p *glusterfsVolumeProvisioner) CreateVolume(gid int) (r *v1.GlusterfsPersistentVolumeSource, size int, volID string, err error) {
  684. var clusterIDs []string
  685. customVolumeName := ""
  686. epServiceName := ""
  687. kubeClient := p.plugin.host.GetKubeClient()
  688. if kubeClient == nil {
  689. return nil, 0, "", fmt.Errorf("failed to get kube client to update endpoint")
  690. }
  691. if len(p.provisionerConfig.customEpNamePrefix) == 0 {
  692. epServiceName = string(p.options.PVC.UID)
  693. } else {
  694. epServiceName = p.provisionerConfig.customEpNamePrefix + "-" + string(p.options.PVC.UID)
  695. }
  696. epNamespace := p.options.PVC.Namespace
  697. endpoint, service, err := p.createOrGetEndpointService(epNamespace, epServiceName, p.options.PVC)
  698. if err != nil {
  699. klog.Errorf("failed to create endpoint/service %v/%v: %v", epNamespace, epServiceName, err)
  700. return nil, 0, "", fmt.Errorf("failed to create endpoint/service %v/%v: %v", epNamespace, epServiceName, err)
  701. }
  702. klog.V(3).Infof("dynamic endpoint %v and service %v ", endpoint, service)
  703. capacity := p.options.PVC.Spec.Resources.Requests[v1.ResourceName(v1.ResourceStorage)]
  704. // GlusterFS/heketi creates volumes in units of GiB.
  705. sz, err := volumehelpers.RoundUpToGiBInt(capacity)
  706. if err != nil {
  707. return nil, 0, "", err
  708. }
  709. klog.V(2).Infof("create volume of size %dGiB", sz)
  710. if p.url == "" {
  711. return nil, 0, "", fmt.Errorf("failed to create glusterfs REST client, REST URL is empty")
  712. }
  713. cli := gcli.NewClient(p.url, p.user, p.secretValue)
  714. if cli == nil {
  715. return nil, 0, "", fmt.Errorf("failed to create glusterfs REST client, REST server authentication failed")
  716. }
  717. if p.provisionerConfig.clusterID != "" {
  718. clusterIDs = dstrings.Split(p.clusterID, ",")
  719. klog.V(4).Infof("provided clusterIDs %v", clusterIDs)
  720. }
  721. if p.provisionerConfig.volumeNamePrefix != "" {
  722. customVolumeName = fmt.Sprintf("%s_%s_%s_%s", p.provisionerConfig.volumeNamePrefix, p.options.PVC.Namespace, p.options.PVC.Name, uuid.NewUUID())
  723. }
  724. gid64 := int64(gid)
  725. snaps := struct {
  726. Enable bool `json:"enable"`
  727. Factor float32 `json:"factor"`
  728. }{
  729. true,
  730. p.provisionerConfig.thinPoolSnapFactor,
  731. }
  732. volumeReq := &gapi.VolumeCreateRequest{Size: sz, Name: customVolumeName, Clusters: clusterIDs, Gid: gid64, Durability: p.volumeType, GlusterVolumeOptions: p.volumeOptions, Snapshot: snaps}
  733. volume, err := cli.VolumeCreate(volumeReq)
  734. if err != nil {
  735. return nil, 0, "", fmt.Errorf("failed to create volume: %v", err)
  736. }
  737. klog.V(1).Infof("volume with size %d and name %s created", volume.Size, volume.Name)
  738. volID = volume.Id
  739. dynamicHostIps, err := getClusterNodes(cli, volume.Cluster)
  740. if err != nil {
  741. return nil, 0, "", fmt.Errorf("failed to get cluster nodes for volume %s: %v", volume, err)
  742. }
  743. addrlist := make([]v1.EndpointAddress, len(dynamicHostIps))
  744. for i, v := range dynamicHostIps {
  745. addrlist[i].IP = v
  746. }
  747. subset := make([]v1.EndpointSubset, 1)
  748. ports := []v1.EndpointPort{{Port: 1, Protocol: "TCP"}}
  749. endpoint.Subsets = subset
  750. endpoint.Subsets[0].Addresses = addrlist
  751. endpoint.Subsets[0].Ports = ports
  752. _, err = kubeClient.CoreV1().Endpoints(epNamespace).Update(context.TODO(), endpoint, metav1.UpdateOptions{})
  753. if err != nil {
  754. deleteErr := cli.VolumeDelete(volume.Id)
  755. if deleteErr != nil {
  756. klog.Errorf("failed to delete volume: %v, manual deletion of the volume required", deleteErr)
  757. }
  758. klog.V(3).Infof("failed to update endpoint, deleting %s", endpoint)
  759. err = kubeClient.CoreV1().Services(epNamespace).Delete(context.TODO(), epServiceName, nil)
  760. if err != nil && errors.IsNotFound(err) {
  761. klog.V(1).Infof("service %s does not exist in namespace %s", epServiceName, epNamespace)
  762. err = nil
  763. }
  764. if err != nil {
  765. klog.Errorf("failed to delete service %s/%s: %v", epNamespace, epServiceName, err)
  766. }
  767. klog.V(1).Infof("service/endpoint: %s/%s deleted successfully", epNamespace, epServiceName)
  768. return nil, 0, "", fmt.Errorf("failed to update endpoint %s: %v", endpoint, err)
  769. }
  770. klog.V(3).Infof("endpoint %s updated successfully", endpoint)
  771. return &v1.GlusterfsPersistentVolumeSource{
  772. EndpointsName: endpoint.Name,
  773. EndpointsNamespace: &epNamespace,
  774. Path: volume.Name,
  775. ReadOnly: false,
  776. }, sz, volID, nil
  777. }
  778. // createOrGetEndpointService() makes sure an endpoint and service
  779. // exist for the given namespace, PVC name, endpoint name
  780. // I.e. the endpoint or service is only created
  781. // if it does not exist yet.
  782. func (p *glusterfsVolumeProvisioner) createOrGetEndpointService(namespace string, epServiceName string, pvc *v1.PersistentVolumeClaim) (endpoint *v1.Endpoints, service *v1.Service, err error) {
  783. pvcNameOrID := ""
  784. if len(pvc.Name) >= 63 {
  785. pvcNameOrID = string(pvc.UID)
  786. } else {
  787. pvcNameOrID = pvc.Name
  788. }
  789. endpoint = &v1.Endpoints{
  790. ObjectMeta: metav1.ObjectMeta{
  791. Namespace: namespace,
  792. Name: epServiceName,
  793. Labels: map[string]string{
  794. "gluster.kubernetes.io/provisioned-for-pvc": pvcNameOrID,
  795. },
  796. },
  797. }
  798. kubeClient := p.plugin.host.GetKubeClient()
  799. if kubeClient == nil {
  800. return nil, nil, fmt.Errorf("failed to get kube client when creating endpoint service")
  801. }
  802. _, err = kubeClient.CoreV1().Endpoints(namespace).Create(context.TODO(), endpoint, metav1.CreateOptions{})
  803. if err != nil && errors.IsAlreadyExists(err) {
  804. klog.V(1).Infof("endpoint %s already exist in namespace %s", endpoint, namespace)
  805. err = nil
  806. }
  807. if err != nil {
  808. klog.Errorf("failed to create endpoint: %v", err)
  809. return nil, nil, fmt.Errorf("failed to create endpoint: %v", err)
  810. }
  811. service = &v1.Service{
  812. ObjectMeta: metav1.ObjectMeta{
  813. Name: epServiceName,
  814. Namespace: namespace,
  815. Labels: map[string]string{
  816. "gluster.kubernetes.io/provisioned-for-pvc": pvcNameOrID,
  817. },
  818. },
  819. Spec: v1.ServiceSpec{
  820. Ports: []v1.ServicePort{
  821. {Protocol: "TCP", Port: 1}}}}
  822. _, err = kubeClient.CoreV1().Services(namespace).Create(context.TODO(), service, metav1.CreateOptions{})
  823. if err != nil && errors.IsAlreadyExists(err) {
  824. klog.V(1).Infof("service %s already exist in namespace %s", service, namespace)
  825. err = nil
  826. }
  827. if err != nil {
  828. klog.Errorf("failed to create service: %v", err)
  829. return nil, nil, fmt.Errorf("error creating service: %v", err)
  830. }
  831. return endpoint, service, nil
  832. }
  833. func (d *glusterfsVolumeDeleter) deleteEndpointService(namespace string, epServiceName string) (err error) {
  834. kubeClient := d.plugin.host.GetKubeClient()
  835. if kubeClient == nil {
  836. return fmt.Errorf("failed to get kube client when deleting endpoint service")
  837. }
  838. err = kubeClient.CoreV1().Services(namespace).Delete(context.TODO(), epServiceName, nil)
  839. if err != nil {
  840. return fmt.Errorf("failed to delete service %s/%s: %v", namespace, epServiceName, err)
  841. }
  842. klog.V(1).Infof("service/endpoint: %s/%s deleted successfully", namespace, epServiceName)
  843. return nil
  844. }
  845. // parseSecret finds a given Secret instance and reads user password from it.
  846. func parseSecret(namespace, secretName string, kubeClient clientset.Interface) (string, error) {
  847. secretMap, err := volutil.GetSecretForPV(namespace, secretName, glusterfsPluginName, kubeClient)
  848. if err != nil {
  849. klog.Errorf("failed to get secret: %s/%s: %v", namespace, secretName, err)
  850. return "", fmt.Errorf("failed to get secret %s/%s: %v", namespace, secretName, err)
  851. }
  852. if len(secretMap) == 0 {
  853. return "", fmt.Errorf("empty secret map")
  854. }
  855. secret := ""
  856. for k, v := range secretMap {
  857. if k == secretKeyName {
  858. return v, nil
  859. }
  860. secret = v
  861. }
  862. // If not found, the last secret in the map wins as done before
  863. return secret, nil
  864. }
  865. // getClusterNodes() returns the cluster nodes of a given cluster
  866. func getClusterNodes(cli *gcli.Client, cluster string) (dynamicHostIps []string, err error) {
  867. clusterinfo, err := cli.ClusterInfo(cluster)
  868. if err != nil {
  869. return nil, fmt.Errorf("failed to get cluster details: %v", err)
  870. }
  871. // For the dynamically provisioned volume, we gather the list of node IPs
  872. // of the cluster on which provisioned volume belongs to, as there can be multiple
  873. // clusters.
  874. for _, node := range clusterinfo.Nodes {
  875. nodeInfo, err := cli.NodeInfo(string(node))
  876. if err != nil {
  877. return nil, fmt.Errorf("failed to get host ipaddress: %v", err)
  878. }
  879. ipaddr := dstrings.Join(nodeInfo.NodeAddRequest.Hostnames.Storage, "")
  880. // IP validates if a string is a valid IP address.
  881. ip := net.ParseIP(ipaddr)
  882. if ip == nil {
  883. return nil, fmt.Errorf("glusterfs server node ip address %s must be a valid IP address, (e.g. 10.9.8.7)", ipaddr)
  884. }
  885. dynamicHostIps = append(dynamicHostIps, ipaddr)
  886. }
  887. klog.V(3).Infof("host list :%v", dynamicHostIps)
  888. if len(dynamicHostIps) == 0 {
  889. return nil, fmt.Errorf("no hosts found: %v", err)
  890. }
  891. return dynamicHostIps, nil
  892. }
  893. // parseClassParameters parses StorageClass parameters.
  894. func parseClassParameters(params map[string]string, kubeClient clientset.Interface) (*provisionerConfig, error) {
  895. var cfg provisionerConfig
  896. var err error
  897. cfg.gidMin = defaultGidMin
  898. cfg.gidMax = defaultGidMax
  899. cfg.customEpNamePrefix = dynamicEpSvcPrefix
  900. authEnabled := true
  901. parseVolumeType := ""
  902. parseVolumeOptions := ""
  903. parseVolumeNamePrefix := ""
  904. parseThinPoolSnapFactor := ""
  905. //thin pool snap factor default to 1.0
  906. cfg.thinPoolSnapFactor = float32(1.0)
  907. for k, v := range params {
  908. switch dstrings.ToLower(k) {
  909. case "resturl":
  910. cfg.url = v
  911. case "restuser":
  912. cfg.user = v
  913. case "restuserkey":
  914. cfg.userKey = v
  915. case "secretname":
  916. cfg.secretName = v
  917. case "secretnamespace":
  918. cfg.secretNamespace = v
  919. case "clusterid":
  920. if len(v) != 0 {
  921. cfg.clusterID = v
  922. }
  923. case "restauthenabled":
  924. authEnabled = dstrings.ToLower(v) == "true"
  925. case "gidmin":
  926. parseGidMin, err := convertGid(v)
  927. if err != nil {
  928. return nil, fmt.Errorf("invalid gidMin value %q for volume plugin %s", k, glusterfsPluginName)
  929. }
  930. if parseGidMin < absoluteGidMin {
  931. return nil, fmt.Errorf("gidMin must be >= %v", absoluteGidMin)
  932. }
  933. if parseGidMin > absoluteGidMax {
  934. return nil, fmt.Errorf("gidMin must be <= %v", absoluteGidMax)
  935. }
  936. cfg.gidMin = parseGidMin
  937. case "gidmax":
  938. parseGidMax, err := convertGid(v)
  939. if err != nil {
  940. return nil, fmt.Errorf("invalid gidMax value %q for volume plugin %s", k, glusterfsPluginName)
  941. }
  942. if parseGidMax < absoluteGidMin {
  943. return nil, fmt.Errorf("gidMax must be >= %v", absoluteGidMin)
  944. }
  945. if parseGidMax > absoluteGidMax {
  946. return nil, fmt.Errorf("gidMax must be <= %v", absoluteGidMax)
  947. }
  948. cfg.gidMax = parseGidMax
  949. case "volumetype":
  950. parseVolumeType = v
  951. case "volumeoptions":
  952. if len(v) != 0 {
  953. parseVolumeOptions = v
  954. }
  955. case "volumenameprefix":
  956. if len(v) != 0 {
  957. parseVolumeNamePrefix = v
  958. }
  959. case "snapfactor":
  960. if len(v) != 0 {
  961. parseThinPoolSnapFactor = v
  962. }
  963. case "customepnameprefix":
  964. // If the string has > 'maxCustomEpNamePrefixLen' chars, the final endpoint name will
  965. // exceed the limitation of 63 chars, so fail if prefix is > 'maxCustomEpNamePrefixLen'
  966. // characters. This is only applicable for 'customepnameprefix' string and default ep name
  967. // string will always pass.
  968. if len(v) <= maxCustomEpNamePrefixLen {
  969. cfg.customEpNamePrefix = v
  970. } else {
  971. return nil, fmt.Errorf("'customepnameprefix' value should be < %d characters", maxCustomEpNamePrefixLen)
  972. }
  973. default:
  974. return nil, fmt.Errorf("invalid option %q for volume plugin %s", k, glusterfsPluginName)
  975. }
  976. }
  977. if len(cfg.url) == 0 {
  978. return nil, fmt.Errorf("StorageClass for provisioner %s must contain 'resturl' parameter", glusterfsPluginName)
  979. }
  980. if len(parseVolumeType) == 0 {
  981. cfg.volumeType = gapi.VolumeDurabilityInfo{Type: gapi.DurabilityReplicate, Replicate: gapi.ReplicaDurability{Replica: replicaCount}}
  982. } else {
  983. parseVolumeTypeInfo := dstrings.Split(parseVolumeType, ":")
  984. switch parseVolumeTypeInfo[0] {
  985. case "replicate":
  986. if len(parseVolumeTypeInfo) >= 2 {
  987. newReplicaCount, err := convertVolumeParam(parseVolumeTypeInfo[1])
  988. if err != nil {
  989. return nil, fmt.Errorf("error parsing volumeType %q: %s", parseVolumeTypeInfo[1], err)
  990. }
  991. cfg.volumeType = gapi.VolumeDurabilityInfo{Type: gapi.DurabilityReplicate, Replicate: gapi.ReplicaDurability{Replica: newReplicaCount}}
  992. } else {
  993. cfg.volumeType = gapi.VolumeDurabilityInfo{Type: gapi.DurabilityReplicate, Replicate: gapi.ReplicaDurability{Replica: replicaCount}}
  994. }
  995. case "disperse":
  996. if len(parseVolumeTypeInfo) >= 3 {
  997. newDisperseData, err := convertVolumeParam(parseVolumeTypeInfo[1])
  998. if err != nil {
  999. return nil, fmt.Errorf("error parsing volumeType %q: %s", parseVolumeTypeInfo[1], err)
  1000. }
  1001. newDisperseRedundancy, err := convertVolumeParam(parseVolumeTypeInfo[2])
  1002. if err != nil {
  1003. return nil, fmt.Errorf("error parsing volumeType %q: %s", parseVolumeTypeInfo[2], err)
  1004. }
  1005. cfg.volumeType = gapi.VolumeDurabilityInfo{Type: gapi.DurabilityEC, Disperse: gapi.DisperseDurability{Data: newDisperseData, Redundancy: newDisperseRedundancy}}
  1006. } else {
  1007. return nil, fmt.Errorf("StorageClass for provisioner %q must have data:redundancy count set for disperse volumes in storage class option '%s'", glusterfsPluginName, "volumetype")
  1008. }
  1009. case "none":
  1010. cfg.volumeType = gapi.VolumeDurabilityInfo{Type: gapi.DurabilityDistributeOnly}
  1011. default:
  1012. return nil, fmt.Errorf("error parsing value for option 'volumetype' for volume plugin %s", glusterfsPluginName)
  1013. }
  1014. }
  1015. if !authEnabled {
  1016. cfg.user = ""
  1017. cfg.secretName = ""
  1018. cfg.secretNamespace = ""
  1019. cfg.userKey = ""
  1020. cfg.secretValue = ""
  1021. }
  1022. if len(cfg.secretName) != 0 || len(cfg.secretNamespace) != 0 {
  1023. // secretName + Namespace has precedence over userKey
  1024. if len(cfg.secretName) != 0 && len(cfg.secretNamespace) != 0 {
  1025. cfg.secretValue, err = parseSecret(cfg.secretNamespace, cfg.secretName, kubeClient)
  1026. if err != nil {
  1027. return nil, err
  1028. }
  1029. } else {
  1030. return nil, fmt.Errorf("StorageClass for provisioner %q must have secretNamespace and secretName either both set or both empty", glusterfsPluginName)
  1031. }
  1032. } else {
  1033. cfg.secretValue = cfg.userKey
  1034. }
  1035. if cfg.gidMin > cfg.gidMax {
  1036. return nil, fmt.Errorf("storageClass for provisioner %q must have gidMax value >= gidMin", glusterfsPluginName)
  1037. }
  1038. if len(parseVolumeOptions) != 0 {
  1039. volOptions := dstrings.Split(parseVolumeOptions, ",")
  1040. if len(volOptions) == 0 {
  1041. return nil, fmt.Errorf("storageClass for provisioner %q must have valid (for e.g., 'client.ssl on') volume option", glusterfsPluginName)
  1042. }
  1043. cfg.volumeOptions = volOptions
  1044. }
  1045. if len(parseVolumeNamePrefix) != 0 {
  1046. if dstrings.Contains(parseVolumeNamePrefix, "_") {
  1047. return nil, fmt.Errorf("storageclass parameter 'volumenameprefix' should not contain '_' in its value")
  1048. }
  1049. cfg.volumeNamePrefix = parseVolumeNamePrefix
  1050. }
  1051. if len(parseThinPoolSnapFactor) != 0 {
  1052. thinPoolSnapFactor, err := strconv.ParseFloat(parseThinPoolSnapFactor, 32)
  1053. if err != nil {
  1054. return nil, fmt.Errorf("failed to convert snapfactor %v to float: %v", parseThinPoolSnapFactor, err)
  1055. }
  1056. if thinPoolSnapFactor < 1.0 || thinPoolSnapFactor > 100.0 {
  1057. return nil, fmt.Errorf("invalid snapshot factor %v, the value must be between 1 to 100", thinPoolSnapFactor)
  1058. }
  1059. cfg.thinPoolSnapFactor = float32(thinPoolSnapFactor)
  1060. }
  1061. return &cfg, nil
  1062. }
  1063. // getVolumeID returns volumeID from the PV or volumename.
  1064. func getVolumeID(pv *v1.PersistentVolume, volumeName string) (string, error) {
  1065. volumeID := ""
  1066. // Get volID from pvspec if available, else fill it from volumename.
  1067. if pv != nil {
  1068. if pv.Annotations[heketiVolIDAnn] != "" {
  1069. volumeID = pv.Annotations[heketiVolIDAnn]
  1070. } else {
  1071. volumeID = dstrings.TrimPrefix(volumeName, volPrefix)
  1072. }
  1073. } else {
  1074. return volumeID, fmt.Errorf("provided PV spec is nil")
  1075. }
  1076. if volumeID == "" {
  1077. return volumeID, fmt.Errorf("volume ID is empty")
  1078. }
  1079. return volumeID, nil
  1080. }
  1081. func (plugin *glusterfsPlugin) ExpandVolumeDevice(spec *volume.Spec, newSize resource.Quantity, oldSize resource.Quantity) (resource.Quantity, error) {
  1082. pvSpec := spec.PersistentVolume.Spec
  1083. volumeName := pvSpec.Glusterfs.Path
  1084. klog.V(2).Infof("received request to expand volume %s", volumeName)
  1085. volumeID, err := getVolumeID(spec.PersistentVolume, volumeName)
  1086. if err != nil {
  1087. return oldSize, fmt.Errorf("failed to get volumeID for volume %s: %v", volumeName, err)
  1088. }
  1089. //Get details of StorageClass.
  1090. class, err := volutil.GetClassForVolume(plugin.host.GetKubeClient(), spec.PersistentVolume)
  1091. if err != nil {
  1092. return oldSize, err
  1093. }
  1094. cfg, err := parseClassParameters(class.Parameters, plugin.host.GetKubeClient())
  1095. if err != nil {
  1096. return oldSize, err
  1097. }
  1098. klog.V(4).Infof("expanding volume: %q", volumeID)
  1099. //Create REST server connection
  1100. cli := gcli.NewClient(cfg.url, cfg.user, cfg.secretValue)
  1101. if cli == nil {
  1102. klog.Errorf("failed to create glusterfs REST client")
  1103. return oldSize, fmt.Errorf("failed to create glusterfs REST client, REST server authentication failed")
  1104. }
  1105. // Find out delta size
  1106. expansionSize := resource.NewScaledQuantity((newSize.Value() - oldSize.Value()), 0)
  1107. expansionSizeGiB := int(volumehelpers.RoundUpToGiB(*expansionSize))
  1108. // Find out requested Size
  1109. requestGiB := volumehelpers.RoundUpToGiB(newSize)
  1110. //Check the existing volume size
  1111. currentVolumeInfo, err := cli.VolumeInfo(volumeID)
  1112. if err != nil {
  1113. klog.Errorf("error when fetching details of volume %s: %v", volumeName, err)
  1114. return oldSize, err
  1115. }
  1116. if int64(currentVolumeInfo.Size) >= requestGiB {
  1117. return newSize, nil
  1118. }
  1119. // Make volume expansion request
  1120. volumeExpandReq := &gapi.VolumeExpandRequest{Size: expansionSizeGiB}
  1121. // Expand the volume
  1122. volumeInfoRes, err := cli.VolumeExpand(volumeID, volumeExpandReq)
  1123. if err != nil {
  1124. klog.Errorf("failed to expand volume %s: %v", volumeName, err)
  1125. return oldSize, err
  1126. }
  1127. klog.V(2).Infof("volume %s expanded to new size %d successfully", volumeName, volumeInfoRes.Size)
  1128. newVolumeSize := resource.MustParse(fmt.Sprintf("%dGi", volumeInfoRes.Size))
  1129. return newVolumeSize, nil
  1130. }