-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Implement a CSI for kubernetes that prepopulates a mounted emptyDir with
a cache
- Loading branch information
Showing
10 changed files
with
575 additions
and
37 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,204 @@ | ||
| package api | ||
|
|
||
| import ( | ||
| "context" | ||
| "fmt" | ||
| "math" | ||
| "os" | ||
| "path" | ||
| "path/filepath" | ||
| "syscall" | ||
|
|
||
| "github.com/container-storage-interface/spec/lib/go/csi" | ||
| "github.com/gadget-inc/dateilager/internal/key" | ||
| "github.com/gadget-inc/dateilager/internal/logger" | ||
| "github.com/gadget-inc/dateilager/pkg/version" | ||
| "github.com/golang/protobuf/ptypes/wrappers" | ||
| "google.golang.org/grpc/codes" | ||
| "google.golang.org/grpc/status" | ||
| ) | ||
|
|
||
| const ( | ||
| DriverName = "com.gadget.dateilager.cached" | ||
| ) | ||
|
|
||
| // GetPluginInfo returns metadata of the plugin | ||
| func (c *Cached) GetPluginInfo(ctx context.Context, req *csi.GetPluginInfoRequest) (*csi.GetPluginInfoResponse, error) { | ||
| resp := &csi.GetPluginInfoResponse{ | ||
| Name: DriverName, | ||
| VendorVersion: version.Version, | ||
| } | ||
|
|
||
| return resp, nil | ||
| } | ||
|
|
||
| // GetPluginCapabilities returns available capabilities of the plugin | ||
| func (c *Cached) GetPluginCapabilities(ctx context.Context, req *csi.GetPluginCapabilitiesRequest) (*csi.GetPluginCapabilitiesResponse, error) { | ||
| resp := &csi.GetPluginCapabilitiesResponse{ | ||
| Capabilities: []*csi.PluginCapability{}, | ||
| } | ||
|
|
||
| return resp, nil | ||
| } | ||
|
|
||
| // Probe returns the health and readiness of the plugin | ||
| func (c *Cached) Probe(ctx context.Context, req *csi.ProbeRequest) (*csi.ProbeResponse, error) { | ||
|
|
||
| return &csi.ProbeResponse{ | ||
| Ready: &wrappers.BoolValue{ | ||
| Value: true, | ||
| }, | ||
| }, nil | ||
| } | ||
|
|
||
| // NodeGetCapabilities returns the supported capabilities of the node server | ||
| // this driver has no capabilities like expansion or staging, because we only use it for node local volumes | ||
| func (c *Cached) NodeGetCapabilities(ctx context.Context, req *csi.NodeGetCapabilitiesRequest) (*csi.NodeGetCapabilitiesResponse, error) { | ||
| nscaps := []*csi.NodeServiceCapability{} | ||
|
|
||
| return &csi.NodeGetCapabilitiesResponse{ | ||
| Capabilities: nscaps, | ||
| }, nil | ||
| } | ||
|
|
||
| // NodeGetInfo returns the supported capabilities of the node server. This | ||
| // Usually, a CSI driver would return some interesting stuff about the node here for the controller to use to place volumes, but because we're only supporting node local volumes, we return something very basic | ||
| func (c *Cached) NodeGetInfo(ctx context.Context, req *csi.NodeGetInfoRequest) (*csi.NodeGetInfoResponse, error) { | ||
| return &csi.NodeGetInfoResponse{ | ||
| NodeId: first(os.Getenv("NODE_NAME"), "dev"), | ||
| MaxVolumesPerNode: 110, | ||
| }, nil | ||
| } | ||
|
|
||
| func (c *Cached) NodePublishVolume(ctx context.Context, req *csi.NodePublishVolumeRequest) (*csi.NodePublishVolumeResponse, error) { | ||
| if req.VolumeId == "" { | ||
| return nil, status.Error(codes.InvalidArgument, "NodePublishVolume Volume ID must be provided") | ||
| } | ||
|
|
||
| if req.TargetPath == "" { | ||
| return nil, status.Error(codes.InvalidArgument, "NodePublishVolume Target Path must be provided") | ||
| } | ||
|
|
||
| if req.VolumeCapability == nil { | ||
| return nil, status.Error(codes.InvalidArgument, "NodePublishVolume Volume Capability must be provided") | ||
| } | ||
|
|
||
| targetPath := req.GetTargetPath() | ||
| volumeID := req.GetVolumeId() | ||
| volumeAttributes := req.GetVolumeContext() | ||
|
|
||
| var cachePath string | ||
| var targetPermissions os.FileMode | ||
|
|
||
| if suffix, exists := volumeAttributes["placeCacheAtPath"]; exists { | ||
| // running in suffix mode, desired outcome: | ||
| // - the mount point is writable by the pod | ||
| // - the cache is mounted at the suffix, and is not writable | ||
| cachePath = path.Join(targetPath, suffix) | ||
| targetPermissions = 0777 | ||
| } else { | ||
| // running in unsuffixed mode, desired outcome: | ||
| // - the mount point *is* the cache, and is not writable by the pod | ||
| cachePath = targetPath | ||
| targetPermissions = 0755 | ||
| } | ||
|
|
||
| if err := os.MkdirAll(targetPath, targetPermissions); err != nil { | ||
| return nil, fmt.Errorf("failed to create target directory %s: %s", targetPath, err) | ||
| } | ||
|
|
||
| if err := os.Chmod(targetPath, targetPermissions); err != nil { | ||
| return nil, fmt.Errorf("failed to change ownership of target directory %s: %s", targetPath, err) | ||
| } | ||
|
|
||
| version, err := c.WriteCache(cachePath) | ||
| if err != nil { | ||
| return nil, err | ||
| } | ||
|
|
||
| logger.Info(ctx, "volume published", key.VolumeID.Field(volumeID), key.TargetPath.Field(targetPath), key.Version.Field(version)) | ||
|
|
||
| return &csi.NodePublishVolumeResponse{}, nil | ||
| } | ||
|
|
||
| func (s *Cached) NodeUnpublishVolume(ctx context.Context, req *csi.NodeUnpublishVolumeRequest) (*csi.NodeUnpublishVolumeResponse, error) { | ||
| if req.VolumeId == "" { | ||
| return nil, status.Error(codes.InvalidArgument, "NodeUnpublishVolume Volume ID must be provided") | ||
| } | ||
|
|
||
| if req.TargetPath == "" { | ||
| return nil, status.Error(codes.InvalidArgument, "NodeUnpublishVolume Target Path must be provided") | ||
| } | ||
|
|
||
| targetPath := req.GetTargetPath() | ||
|
|
||
| // Clean up directory | ||
| if err := os.RemoveAll(targetPath); err != nil { | ||
| return nil, fmt.Errorf("failed to remove directory %s: %s", targetPath, err) | ||
| } | ||
|
|
||
| logger.Info(ctx, "volume unpublished and data removed", key.TargetPath.Field(targetPath)) | ||
| return &csi.NodeUnpublishVolumeResponse{}, nil | ||
| } | ||
|
|
||
| // NodeGetVolumeStats returns the volume capacity statistics available for the given volume. | ||
| func (c *Cached) NodeGetVolumeStats(ctx context.Context, req *csi.NodeGetVolumeStatsRequest) (*csi.NodeGetVolumeStatsResponse, error) { | ||
| if req.VolumeId == "" { | ||
| return nil, status.Error(codes.InvalidArgument, "NodeGetVolumeStats Volume ID must be provided") | ||
| } | ||
|
|
||
| volumePath := req.VolumePath | ||
| if volumePath == "" { | ||
| return nil, status.Error(codes.InvalidArgument, "NodeGetVolumeStats Volume Path must be provided") | ||
| } | ||
|
|
||
| usedBytes, err := getFolderSize(volumePath) | ||
| if err != nil { | ||
| return nil, status.Errorf(codes.Internal, "failed to retrieve used size statistics for volume path %q: %s", volumePath, err) | ||
| } | ||
|
|
||
| var stat syscall.Statfs_t | ||
| err = syscall.Statfs(volumePath, &stat) | ||
| if err != nil { | ||
| return nil, status.Errorf(codes.Internal, "failed to retrieve total size statistics for volume path %q: %s", volumePath, err) | ||
| } | ||
|
|
||
| // Calculate free space in bytes | ||
| freeBytes := stat.Bavail * uint64(stat.Bsize) | ||
| if freeBytes > math.MaxInt64 { | ||
| return nil, status.Errorf(codes.Internal, "total size statistics for volume path too big for int64: %d", freeBytes) | ||
| } | ||
| signedFreeBytes := int64(freeBytes) | ||
|
|
||
| return &csi.NodeGetVolumeStatsResponse{ | ||
| Usage: []*csi.VolumeUsage{ | ||
| { | ||
| Available: signedFreeBytes, | ||
| Total: signedFreeBytes + usedBytes, | ||
| Used: usedBytes, | ||
| Unit: csi.VolumeUsage_BYTES, | ||
| }, | ||
| }, | ||
| }, nil | ||
| } | ||
|
|
||
| func first(one, two string) string { | ||
| if one == "" { | ||
| return two | ||
| } | ||
| return one | ||
| } | ||
|
|
||
| func getFolderSize(path string) (int64, error) { | ||
| var totalSize int64 | ||
| err := filepath.Walk(path, func(_ string, info os.FileInfo, err error) error { | ||
| if err != nil { | ||
| return err | ||
| } | ||
| if !info.IsDir() { | ||
| totalSize += info.Size() | ||
| } | ||
| return nil | ||
| }) | ||
| return totalSize, err | ||
| } |
Oops, something went wrong.