node.go

/*
Package csi is CSI driver interface for OSD
Copyright 2017 Portworx

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

 http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
*/
package csi

import (
"fmt"
"os"
"strings"

 csi "github.com/container-storage-interface/spec/lib/go/csi"
"github.com/libopenstorage/openstorage/api"
"github.com/libopenstorage/openstorage/pkg/grpcutil"
"github.com/portworx/kvdb"
"golang.org/x/net/context"
"google.golang.org/grpc"
"google.golang.org/grpc/codes"
"google.golang.org/grpc/status"
)

var (
ephemeralDenyList= []string{
api.SpecPriorityAlias,
api.SpecPriority,
api.SpecSticky,
api.SpecScale,
 }
)

func (s*OsdCsiServer) NodeGetInfo(
ctx context.Context,
req*csi.NodeGetInfoRequest,
) (*csi.NodeGetInfoResponse, error) {

clus, err:=s.cluster.Enumerate()
iferr!=nil {
returnnil, status.Errorf(codes.Internal, "Unable to Enumerate cluster: %s", err)
 }

node, err:=s.cluster.Inspect(clus.NodeId)
iferr!=nil {
returnnil, status.Errorf(codes.Internal, "Unable to Inspect node %s: %s", clus.NodeId, err)
 }

topology:=&csi.Topology{}
ifnode.SchedulerTopology!=nil&&len(node.SchedulerTopology.Labels) >0 {
topology.Segments=make(map[string]string)
fork, v:=rangenode.SchedulerTopology.Labels {
topology.Segments[k] =v
 }
 }

return&csi.NodeGetInfoResponse{
NodeId: clus.NodeId,
AccessibleTopology: topology,
 }, nil
}

// NodePublishVolume is a CSI API call which mounts the volume on the specified
// target path on the node.
//
// TODO: Support READ ONLY Mounts
func (s*OsdCsiServer) NodePublishVolume(
ctx context.Context,
req*csi.NodePublishVolumeRequest,
) (*csi.NodePublishVolumeResponse, error) {
volumeId:=req.GetVolumeId()
targetPath:=req.GetTargetPath()

clogger.WithContext(ctx).Infof("csi.NodePublishVolume request received. VolumeID: %s, TargetPath: %s", volumeId, targetPath)

// Check arguments
iflen(volumeId) ==0 {
returnnil, status.Error(codes.InvalidArgument, "Volume id must be provided")
 }
iflen(targetPath) ==0 {
returnnil, status.Error(codes.InvalidArgument, "Target path must be provided")
 }
ifreq.GetVolumeCapability() ==nil||req.GetVolumeCapability().GetAccessMode() ==nil||
req.GetVolumeCapability().GetAccessMode().Mode==csi.VolumeCapability_AccessMode_UNKNOWN {
returnnil, status.Error(codes.InvalidArgument, "Volume access mode must be provided")
 }

// Ensure target location is created correctly
isBlockAccessType:=false
ifreq.GetVolumeCapability().GetBlock() !=nil {
isBlockAccessType=true
 }
iferr:=ensureMountPathCreated(ctx, targetPath, isBlockAccessType); err!=nil {
returnnil, status.Errorf(
codes.Aborted,
"Failed to use target location %s: %s",
targetPath,
err.Error())
 }

// Get grpc connection
conn, err:=s.getConn()
iferr!=nil {
returnnil, status.Errorf(
codes.Unavailable,
"Unable to connect to SDK server: %v", err)
 }

// Get secret if any was passed
ctx=s.setupContext(ctx, req.GetSecrets())
ctx, cancel:=grpcutil.WithDefaultTimeout(ctx)
defercancel()

// Check if block device
driverType:=s.driver.Type()
ifdriverType!=api.DriverType_DRIVER_TYPE_BLOCK&&
req.GetVolumeCapability().GetBlock() !=nil {
returnnil, status.Errorf(codes.InvalidArgument, "Trying to attach as block a non block device")
 }

// Gather volume attributes
spec, locator, _, err:=s.specHandler.SpecFromOpts(req.GetVolumeContext())
iferr!=nil {
returnnil, status.Errorf(
codes.InvalidArgument,
"Invalid volume attributes: %#v",
req.GetVolumeContext())
 }

// Get volume encryption info from req.Secrets
driverOpts:=s.addEncryptionInfoToLabels(make(map[string]string), req.GetSecrets())

// Parse storage class 'mountOptions' flags from CSI req
// flags from 'mountOptions' will be used as the only source of truth for Pure volumes upon mounting
ifreq.GetVolumeCapability() !=nil&&req.GetVolumeCapability().GetMount() !=nil {
mountFlags:=strings.Join(req.GetVolumeCapability().GetMount().GetMountFlags(), ",")
ifmountFlags!="" {
driverOpts[api.SpecCSIMountOptions] =mountFlags
 }
 }

// can use either spec.Ephemeral or VolumeContext label
ifreq.GetVolumeContext()["csi.storage.k8s.io/ephemeral"] =="true"||spec.Ephemeral {
if!s.allowInlineVolumes {
returnnil, status.Error(codes.InvalidArgument, "CSI ephemeral inline volumes are disabled on this cluster")
 }

iferr:=validateEphemeralVolumeAttributes(req.GetVolumeContext()); err!=nil {
returnnil, status.Error(codes.InvalidArgument, err.Error())
 }

spec.Ephemeral=true
volumes:=api.NewOpenStorageVolumeClient(conn)
resp, err:=volumes.Create(ctx, &api.SdkVolumeCreateRequest{
Name: volumeId,
Spec: spec,
Labels: locator.GetVolumeLabels(),
 })
iferr!=nil {
returnnil, err
 }
volumeId=resp.VolumeId
 }

// prepare for mount/attaching
opts:=&api.SdkVolumeAttachOptions{
SecretName: spec.GetPassphrase(),
 }
mounts:=api.NewOpenStorageMountAttachClient(conn)
ifdriverType==api.DriverType_DRIVER_TYPE_BLOCK {
// attach is assumed to be idempotent
// attach is assumed to return the same DevicePath on each call
if_, err=mounts.Attach(ctx, &api.SdkVolumeAttachRequest{
VolumeId: volumeId,
Options: opts,
DriverOptions: driverOpts,
 }); err!=nil {
ifspec.Ephemeral {
clogger.WithContext(ctx).Errorf("Failed to attach ephemeral volume %s: %v", volumeId, err.Error())
s.cleanupEphemeral(ctx, conn, volumeId, false)
 }
returnnil, adjustFinalErrors(err)
 }
 }

// for volumes with mount access type just mount volume onto the path
if_, err:=mounts.Mount(ctx, &api.SdkVolumeMountRequest{
VolumeId: volumeId,
MountPath: targetPath,
Options: opts,
DriverOptions: driverOpts,
 }); err!=nil {
ifspec.Ephemeral {
clogger.WithContext(ctx).Errorf("Failed to mount ephemeral volume %s: %v", volumeId, err.Error())
s.cleanupEphemeral(ctx, conn, volumeId, true)
 }
returnnil, adjustFinalErrors(err)
 }

clogger.WithContext(ctx).Infof("CSI Volume %s mounted on %s",
volumeId,
req.GetTargetPath())

return&csi.NodePublishVolumeResponse{}, nil
}

// NodeUnpublishVolume is a CSI API call which unmounts the volume.
func (s*OsdCsiServer) NodeUnpublishVolume(
ctx context.Context,
req*csi.NodeUnpublishVolumeRequest,
) (*csi.NodeUnpublishVolumeResponse, error) {
volumeId:=req.GetVolumeId()
targetPath:=req.GetTargetPath()

clogger.WithContext(ctx).Infof("csi.NodeUnpublishVolume request received. VolumeID: %s, TargetPath: %s", volumeId, targetPath)

// Check arguments
iflen(volumeId) ==0 {
returnnil, status.Error(codes.InvalidArgument, "Volume id must be provided")
 }
iflen(targetPath) ==0 {
returnnil, status.Error(codes.InvalidArgument, "Target path must be provided")
 }

// Get volume information
vols, err:=s.driver.Enumerate(&api.VolumeLocator{VolumeIds: []string{req.GetVolumeId()}}, nil)
iferr!=nil||len(vols) <1 {
iferr==kvdb.ErrNotFound {
clogger.WithContext(ctx).Tracef("Volume %s was deleted or cannot be found: %s", req.GetVolumeId(), err.Error())
return&csi.NodeUnpublishVolumeResponse{}, nil
 } elseiferr!=nil {
returnnil, status.Errorf(codes.NotFound, "Volume id %s not found: %s",
req.GetVolumeId(),
err.Error())
 } else {
clogger.WithContext(ctx).Tracef("Volume %s was deleted or cannot be found", req.GetVolumeId())
return&csi.NodeUnpublishVolumeResponse{}, nil
 }
 }

iferr=s.driver.Unmount(ctx, req.GetVolumeId(), req.GetTargetPath(), nil); err!=nil {
clogger.WithContext(ctx).Infof("unable to unmount volume %s onto %s: %s",
req.GetVolumeId(),
req.GetTargetPath(),
err.Error(),
 )
 }

ifs.driver.Type() ==api.DriverType_DRIVER_TYPE_BLOCK {
iferr=s.driver.Detach(ctx, volumeId, nil); err!=nil {
returnnil, status.Errorf(
codes.Canceled,
"Unable to detach volume: %s",
err.Error())
 }
 }

// Attempt to remove volume path
// Kubernetes handles this after NodeUnpublishVolume finishes, but this allows for cross-CO compatibility
iferr:=os.Remove(req.GetTargetPath()); err!=nil&&!os.IsNotExist(err) {
clogger.WithContext(ctx).Tracef("Failed to delete mount path %s: %s", targetPath, err.Error())
 }

// Return error to Kubelet if mount path still exists to force a retry
if_, err:=os.Stat(targetPath); !os.IsNotExist(err) {
returnnil, status.Errorf(
codes.Canceled,
"Mount path still exists: %s",
targetPath)
 }

clogger.WithContext(ctx).Infof("CSI Volume %s unmounted from path %s", volumeId, targetPath)

return&csi.NodeUnpublishVolumeResponse{}, nil
}

// NodeGetCapabilities is a CSI API function which seems to be setup for
// future patches
func (s*OsdCsiServer) NodeGetCapabilities(
ctx context.Context,
req*csi.NodeGetCapabilitiesRequest,
) (*csi.NodeGetCapabilitiesResponse, error) {
clogger.WithContext(ctx).Debugf("csi.NodeGetCapabilities request received")

caps:= []csi.NodeServiceCapability_RPC_Type{
// Getting volume stats for volume health monitoring
csi.NodeServiceCapability_RPC_GET_VOLUME_STATS,
// Indicates that the Node service can report volume conditions.
csi.NodeServiceCapability_RPC_VOLUME_CONDITION,
 }

varserviceCapabilities []*csi.NodeServiceCapability
for_, cap:=rangecaps {
serviceCapabilities=append(serviceCapabilities, &csi.NodeServiceCapability{
Type: &csi.NodeServiceCapability_Rpc{
Rpc: &csi.NodeServiceCapability_RPC{
Type: cap,
 },
 },
 })
 }

return&csi.NodeGetCapabilitiesResponse{
Capabilities: serviceCapabilities,
 }, nil
}

funcgetVolumeCondition(vol*api.Volume) *csi.VolumeCondition {
condition:=&csi.VolumeCondition{}
ifvol.Status!=api.VolumeStatus_VOLUME_STATUS_UP {
condition.Abnormal=true
 }

switchvol.Status {
caseapi.VolumeStatus_VOLUME_STATUS_UP:
condition.Message="Volume status is up"

caseapi.VolumeStatus_VOLUME_STATUS_NOT_PRESENT:
condition.Message="Volume status is not present"

caseapi.VolumeStatus_VOLUME_STATUS_DOWN:
condition.Message="Volume status is down"

caseapi.VolumeStatus_VOLUME_STATUS_DEGRADED:
condition.Message="Volume status is degraded"

default:
condition.Message="Volume status is unknown"
 }

returncondition
}

// NodeGetVolumeStats get volume stats for a given node.
// This function skips auth and directly hits the driver as it is read-only
// and only exposed via the CSI unix domain socket. If a secrets field is added
// in csi.NodeGetVolumeStatsRequest, we can update this to hit the SDK and use auth.
func (s*OsdCsiServer) NodeGetVolumeStats(ctx context.Context, req*csi.NodeGetVolumeStatsRequest) (*csi.NodeGetVolumeStatsResponse, error) {
clogger.WithContext(ctx).Debugf("NodeGetVolumeStats request received. VolumeID: %s, VolumePath: %s", req.GetVolumeId(), req.GetVolumePath())

// Check arguments
id:=req.GetVolumeId()
iflen(id) ==0 {
returnnil, status.Error(codes.InvalidArgument, "Volume id must be provided")
 }
path:=req.GetVolumePath()
iflen(path) ==0 {
returnnil, status.Error(codes.InvalidArgument, "Volume path must be provided")
 }

// Driver inspect as NodeGetVolumeStatsRequest does not support secrets
vol, err:=s.driverGetVolume(ctx, req.GetVolumeId())
iferr!=nil {
returnnil, err
 }

varattachPathMatchbool
sharedPath:=fmt.Sprintf("%s/%s", api.SharedVolExportPrefix, id)
for_, attachPath:=rangevol.AttachPath {
ifattachPath==path||attachPath==sharedPath {
attachPathMatch=true
 }
 }
if!attachPathMatch {
returnnil, status.Errorf(codes.NotFound, "Volume %s not mounted on path %s", id, path)
 }

// Define volume usage
total:=int64(vol.Spec.Size)
used:=int64(vol.Usage)
usage:=&csi.VolumeUsage{
Available: total-used,
Total: total,
Used: used,
Unit: csi.VolumeUsage_BYTES,
 }

// Define volume condition
return&csi.NodeGetVolumeStatsResponse{
Usage: []*csi.VolumeUsage{
usage,
 },
VolumeCondition: getVolumeCondition(vol),
 }, nil
}

// cleanupEphemeral detaches and deletes an ephemeral volume if either attach or mount fails
func (s*OsdCsiServer) cleanupEphemeral(ctx context.Context, conn*grpc.ClientConn, volumeIdstring, detachbool) {
ifdetach {
mounts:=api.NewOpenStorageMountAttachClient(conn)
if_, err:=mounts.Detach(ctx, &api.SdkVolumeDetachRequest{
VolumeId: volumeId,
 }); err!=nil {
clogger.WithContext(ctx).Errorf("Failed to detach ephemeral volume %s during cleanup: %v", volumeId, err.Error())
return
 }
 }
volumes:=api.NewOpenStorageVolumeClient(conn)
if_, err:=volumes.Delete(ctx, &api.SdkVolumeDeleteRequest{
VolumeId: volumeId,
 }); err!=nil {
clogger.WithContext(ctx).Errorf("Failed to delete ephemeral volume %s during cleanup: %v", volumeId, err.Error())
 }
}

funcensureMountPathCreated(ctx context.Context, targetPathstring, isBlockbool) error {
// Check if targetpath exists
fileInfo, err:=os.Lstat(targetPath)
iferr!=nil&&os.IsNotExist(err) {
// Create if does not exist
// 1. Block - create targetPath file
// 2. Mount - create targetpath directory
ifisBlock {
iferr=makeFile(ctx, targetPath); err!=nil {
returnerr
 }
 } else {
iferr=makeDir(targetPath); err!=nil {
returnerr
 }
 }

returnnil
 } elseiferr!=nil {
returnfmt.Errorf(
"unknown error while verifying target location %s: %s",
targetPath,
err.Error())
 }

// Check for directory or file.
// 1. Block - should be file
// 2. Mount - should be directory
ifisBlock {
iffileInfo.IsDir() {
returnfmt.Errorf("Target location %s is not a file", targetPath)
 }
 } else {
if!fileInfo.IsDir() {
returnfmt.Errorf("Target location %s is not a directory", targetPath)
 }
 }

returnnil
}

funcvalidateEphemeralVolumeAttributes(volumeAttributesmap[string]string) error {
forattr:=rangevolumeAttributes {
for_, deny:=rangeephemeralDenyList {
ifattr==deny {
returnfmt.Errorf("invalid ephemeral volume attribute provided. "+
"Volume attributes %v are not allowed for ephemeral volumes", ephemeralDenyList)
 }
 }
 }

returnnil
}

funcmakeFile(ctx context.Context, pathnamestring) error {
f, err:=os.OpenFile(pathname, os.O_CREATE, os.FileMode(0644))
deferfunc() {
err:=f.Close()
iferr!=nil {
clogger.WithContext(ctx).Warnf("failed to close file: %s", err.Error())
 }
 }()
iferr!=nil {
if!os.IsExist(err) {
returnfmt.Errorf("failed to create block file: %s", err.Error())
 }
 }

returnnil
}

funcmakeDir(targetPathstring) error {
err:=os.MkdirAll(targetPath, 0750)
iferr!=nil {
returnfmt.Errorf(
"failed to create target path %s: %s",
targetPath,
err.Error())
 }

returnnil
}