@@ -17,24 +17,23 @@ limitations under the License.
1717package hostpath
1818
1919import (
20- "context"
2120 "errors"
2221 "fmt"
2322 "io"
2423 "io/ioutil"
2524 "os"
25+ "os/exec"
2626 "path/filepath"
27+ "regexp"
2728 "sort"
2829 "strings"
2930
3031 "github.com/golang/glog"
3132 timestamp "github.com/golang/protobuf/ptypes/timestamp"
3233 "google.golang.org/grpc/codes"
3334 "google.golang.org/grpc/status"
34- v1 "k8s.io/api/core/v1"
35- metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
3635 "k8s.io/client-go/kubernetes"
37- "k8s.io/client-go/tools/clientcmd "
36+ fs "k8s.io/kubernetes/pkg/volume/util/fs "
3837 "k8s.io/kubernetes/pkg/volume/util/volumepathhandler"
3938 utilexec "k8s.io/utils/exec"
4039)
@@ -129,26 +128,13 @@ func NewHostPathDriver(driverName, nodeID, endpoint string, ephemeral bool, maxV
129128 glog .Infof ("Driver: %v " , driverName )
130129 glog .Infof ("Version: %s" , vendorVersion )
131130
132- config , err := clientcmd .BuildConfigFromFlags ("" , kubeconfig )
133- if err != nil {
134- fmt .Printf ("Failed to build kubeconfig: %s" , err .Error ())
135- os .Exit (1 )
136- }
137-
138- clientset , err := kubernetes .NewForConfig (config )
139- if err != nil {
140- fmt .Printf ("Faild to create k8s client: %s" , err .Error ())
141- os .Exit (1 )
142- }
143-
144131 return & hostPath {
145132 name : driverName ,
146133 version : vendorVersion ,
147134 nodeID : nodeID ,
148135 endpoint : endpoint ,
149136 ephemeral : ephemeral ,
150137 maxVolumesPerNode : maxVolumesPerNode ,
151- kclient : clientset ,
152138 }, nil
153139}
154140
@@ -181,10 +167,12 @@ func discoverExistingSnapshots() {
181167 }
182168}
183169
184- func (hp * hostPath ) Run () {
185- // Initialize existing volumes
186- hp .initExistingVolumes ()
170+ func (hp * hostPath ) Run () error {
187171 // Create GRPC servers
172+ if err := initialExistingVolumes (); err != nil {
173+ return err
174+ }
175+
188176 hp .ids = NewIdentityServer (hp .name , hp .version )
189177 hp .ns = NewNodeServer (hp .nodeID , hp .ephemeral , hp .maxVolumesPerNode )
190178 hp .cs = NewControllerServer (hp .ephemeral , hp .nodeID )
@@ -193,32 +181,8 @@ func (hp *hostPath) Run() {
193181 s := NewNonBlockingGRPCServer ()
194182 s .Start (hp .endpoint , hp .ids , hp .cs , hp .ns )
195183 s .Wait ()
196- }
197-
198- func (hp * hostPath ) initExistingVolumes () {
199- pvList , err := hp .kclient .CoreV1 ().PersistentVolumes ().List (context .Background (), metav1.ListOptions {})
200- if err != nil {
201- glog .V (3 ).Infof ("Failed to get pv: %s" , err .Error ())
202- os .Exit (1 )
203- }
204-
205- for _ , pv := range pvList .Items {
206- volAccessType := mountAccess
207- if * pv .Spec .VolumeMode == v1 .PersistentVolumeBlock {
208- volAccessType = blockAccess
209- }
210-
211- hostPathVolumes [pv .Spec .CSI .VolumeHandle ] = hostPathVolume {
212- VolName : pv .Name ,
213- VolID : pv .Spec .CSI .VolumeHandle ,
214- VolSize : pv .Spec .Capacity .Storage ().Value (),
215- VolPath : getVolumePath (pv .Spec .CSI .VolumeHandle ),
216- VolAccessType : volAccessType ,
217- }
218- }
219184
220- glog .V (3 ).Infof ("The existing volume list is: %+v\n " , hostPathVolumes )
221- return
185+ return nil
222186}
223187
224188func getVolumeByID (volumeID string ) (hostPathVolume , error ) {
@@ -463,3 +427,85 @@ func getSortedVolumeIDs() []string {
463427 sort .Strings (ids )
464428 return ids
465429}
430+
431+ func filterVolumeName (targetPath string ) string {
432+ pathItems := strings .Split (targetPath , "kubernetes.io~csi/" )
433+ if len (pathItems ) < 2 {
434+ return ""
435+ }
436+
437+ return strings .TrimSuffix (pathItems [1 ], "/mount" )
438+ }
439+
440+ func filterVolumeID (sourcePath string ) string {
441+ volumeSourcePathRegex := regexp .MustCompile (`\[(.*)\]` )
442+ volumeSP := string (volumeSourcePathRegex .Find ([]byte (sourcePath )))
443+ if volumeSP == "" {
444+ return ""
445+ }
446+
447+ return strings .TrimSuffix (strings .TrimPrefix (volumeSP , "[/var/lib/csi-hostpath-data/" ), "]" )
448+ }
449+
450+ func parseVolumeInfo (volume MountPointInfo ) (* hostPathVolume , error ) {
451+ volumeName := filterVolumeName (volume .Target )
452+ volumeID := filterVolumeID (volume .Source )
453+ sourcePath := getSourcePath (volumeID )
454+ _ , fscapacity , _ , _ , _ , _ , err := fs .FsInfo (sourcePath )
455+ if err != nil {
456+ return nil , fmt .Errorf ("failed to get capacity info: %+v" , err )
457+ }
458+
459+ hp := hostPathVolume {
460+ VolName : volumeName ,
461+ VolID : volumeID ,
462+ VolSize : fscapacity ,
463+ VolPath : getVolumePath (volumeID ),
464+ VolAccessType : mountAccess ,
465+ }
466+
467+ return & hp , nil
468+ }
469+
470+ func initialExistingVolumes () error {
471+ cmdPath := locateCommandPath ("findmnt" )
472+ out , err := exec .Command (cmdPath , "--json" ).CombinedOutput ()
473+ if err != nil {
474+ glog .V (3 ).Infof ("failed to execute command: %+v" , cmdPath )
475+ return err
476+ }
477+
478+ if len (out ) < 1 {
479+ return fmt .Errorf ("mount point info is nil" )
480+ }
481+
482+ mountInfos , err := parseMountInfo ([]byte (out ))
483+ if err != nil {
484+ return fmt .Errorf ("failed to parse the mount infos: %+v" , err )
485+ }
486+
487+ mountInfosOfPod := MountPointInfo {}
488+ for _ , mountInfo := range mountInfos {
489+ if mountInfo .Target == podVolumeTargetPath {
490+ mountInfosOfPod = mountInfo
491+ break
492+ }
493+ }
494+
495+ // getting existing volumes based on the mount point infos.
496+ // It's a temporary solution to recall volumes.
497+ for _ , pv := range mountInfosOfPod .ContainerFileSystem {
498+ if ! strings .Contains (pv .Target , csiSignOfVolumeTargetPath ) {
499+ continue
500+ }
501+
502+ hp , err := parseVolumeInfo (pv )
503+ if err != nil {
504+ return err
505+ }
506+
507+ hostPathVolumes [hp .VolID ] = * hp
508+ }
509+
510+ return nil
511+ }
0 commit comments