5408148111

Committed 29 Jun 2023 03:26AM UTC coverage: 82.888%. Remained the same

Build # 5408148111

Build Type

Pull #1894

github

Committed by

andyzhangx

Commit Message

chore: upgrade azure lib

Pull Request Pull Request #1894: fix: PerformancePlus setting issue

Run Details

4 of 4 new or added lines in 2 files covered. (100.0%)

2824 of 3407 relevant lines covered (82.89%)

6.16 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

85.81

/pkg/azuredisk/azuredisk.go

/*
Copyright 2017 The Kubernetes Authors.

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 azuredisk

import (
        "context"
        "fmt"
        "reflect"
        "strconv"
        "strings"
        "time"

        "github.com/Azure/azure-sdk-for-go/services/compute/mgmt/2022-08-01/compute"
        "github.com/container-storage-interface/spec/lib/go/csi"

        "google.golang.org/grpc/codes"
        "google.golang.org/grpc/status"

        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
        clientset "k8s.io/client-go/kubernetes"
        "k8s.io/klog/v2"
        "k8s.io/kubernetes/pkg/volume/util/hostutil"
        "k8s.io/mount-utils"

        consts "sigs.k8s.io/azuredisk-csi-driver/pkg/azureconstants"
        "sigs.k8s.io/azuredisk-csi-driver/pkg/azureutils"
        csicommon "sigs.k8s.io/azuredisk-csi-driver/pkg/csi-common"
        "sigs.k8s.io/azuredisk-csi-driver/pkg/mounter"
        "sigs.k8s.io/azuredisk-csi-driver/pkg/optimization"
        volumehelper "sigs.k8s.io/azuredisk-csi-driver/pkg/util"
        azcache "sigs.k8s.io/cloud-provider-azure/pkg/cache"
        azurecloudconsts "sigs.k8s.io/cloud-provider-azure/pkg/consts"
        azure "sigs.k8s.io/cloud-provider-azure/pkg/provider"
)

// DriverOptions defines driver parameters specified in driver deployment
type DriverOptions struct {
        NodeID                       string
        DriverName                   string
        VolumeAttachLimit            int64
        EnablePerfOptimization       bool
        CloudConfigSecretName        string
        CloudConfigSecretNamespace   string
        CustomUserAgent              string
        UserAgentSuffix              string
        UseCSIProxyGAInterface       bool
        EnableDiskOnlineResize       bool
        AllowEmptyCloudConfig        bool
        EnableAsyncAttach            bool
        EnableListVolumes            bool
        EnableListSnapshots          bool
        SupportZone                  bool
        GetNodeInfoFromLabels        bool
        EnableDiskCapacityCheck      bool
        DisableUpdateCache           bool
        EnableTrafficManager         bool
        TrafficManagerPort           int64
        AttachDetachInitialDelayInMs int64
        VMSSCacheTTLInSeconds        int64
        VMType                       string
        EnableWindowsHostProcess     bool
        GetNodeIDFromIMDS            bool
}

// CSIDriver defines the interface for a CSI driver.
type CSIDriver interface {
        csi.ControllerServer
        csi.NodeServer
        csi.IdentityServer

        Run(endpoint, kubeconfig string, disableAVSetNodes, testMode bool)
}

type hostUtil interface {
        PathIsDevice(string) (bool, error)
}

// DriverCore contains fields common to both the V1 and V2 driver, and implements all interfaces of CSI drivers
type DriverCore struct {
        csicommon.CSIDriver
        perfOptimizationEnabled      bool
        cloudConfigSecretName        string
        cloudConfigSecretNamespace   string
        customUserAgent              string
        userAgentSuffix              string
        kubeconfig                   string
        cloud                        *azure.Cloud
        mounter                      *mount.SafeFormatAndMount
        deviceHelper                 optimization.Interface
        nodeInfo                     *optimization.NodeInfo
        ioHandler                    azureutils.IOHandler
        hostUtil                     hostUtil
        useCSIProxyGAInterface       bool
        enableDiskOnlineResize       bool
        allowEmptyCloudConfig        bool
        enableAsyncAttach            bool
        enableListVolumes            bool
        enableListSnapshots          bool
        supportZone                  bool
        getNodeInfoFromLabels        bool
        enableDiskCapacityCheck      bool
        disableUpdateCache           bool
        enableTrafficManager         bool
        trafficManagerPort           int64
        vmssCacheTTLInSeconds        int64
        attachDetachInitialDelayInMs int64
        vmType                       string
        enableWindowsHostProcess     bool
        getNodeIDFromIMDS            bool
}

// Driver is the v1 implementation of the Azure Disk CSI Driver.
type Driver struct {
        DriverCore
        volumeLocks *volumehelper.VolumeLocks
        // a timed cache GetDisk throttling
        getDiskThrottlingCache azcache.Resource
}

// newDriverV1 Creates a NewCSIDriver object. Assumes vendor version is equal to driver version &
// does not support optional driver plugin info manifest field. Refer to CSI spec for more details.
func newDriverV1(options *DriverOptions) *Driver {
        driver := Driver{}
        driver.Name = options.DriverName
        driver.Version = driverVersion
        driver.NodeID = options.NodeID
        driver.VolumeAttachLimit = options.VolumeAttachLimit
        driver.perfOptimizationEnabled = options.EnablePerfOptimization
        driver.cloudConfigSecretName = options.CloudConfigSecretName
        driver.cloudConfigSecretNamespace = options.CloudConfigSecretNamespace
        driver.customUserAgent = options.CustomUserAgent
        driver.userAgentSuffix = options.UserAgentSuffix
        driver.useCSIProxyGAInterface = options.UseCSIProxyGAInterface
        driver.enableDiskOnlineResize = options.EnableDiskOnlineResize
        driver.allowEmptyCloudConfig = options.AllowEmptyCloudConfig
        driver.enableAsyncAttach = options.EnableAsyncAttach
        driver.enableListVolumes = options.EnableListVolumes
        driver.enableListSnapshots = options.EnableListVolumes
        driver.supportZone = options.SupportZone
        driver.getNodeInfoFromLabels = options.GetNodeInfoFromLabels
        driver.enableDiskCapacityCheck = options.EnableDiskCapacityCheck
        driver.disableUpdateCache = options.DisableUpdateCache
        driver.attachDetachInitialDelayInMs = options.AttachDetachInitialDelayInMs
        driver.enableTrafficManager = options.EnableTrafficManager
        driver.trafficManagerPort = options.TrafficManagerPort
        driver.vmssCacheTTLInSeconds = options.VMSSCacheTTLInSeconds
        driver.vmType = options.VMType
        driver.enableWindowsHostProcess = options.EnableWindowsHostProcess
        driver.getNodeIDFromIMDS = options.GetNodeIDFromIMDS
        driver.volumeLocks = volumehelper.NewVolumeLocks()
        driver.ioHandler = azureutils.NewOSIOHandler()
        driver.hostUtil = hostutil.NewHostUtil()

        topologyKey = fmt.Sprintf("topology.%s/zone", driver.Name)

        cache, err := azcache.NewTimedCache(5*time.Minute, func(key string) (interface{}, error) {
                return nil, nil
        }, false)
        if err != nil {
                klog.Fatalf("%v", err)
        }
        driver.getDiskThrottlingCache = cache
        return &driver
}

// Run driver initialization
func (d *Driver) Run(endpoint, kubeconfig string, disableAVSetNodes, testingMock bool) {
        versionMeta, err := GetVersionYAML(d.Name)
        if err != nil {
                klog.Fatalf("%v", err)
        }
        klog.Infof("\nDRIVER INFORMATION:\n-------------------\n%s\n\nStreaming logs below:", versionMeta)

        userAgent := GetUserAgent(d.Name, d.customUserAgent, d.userAgentSuffix)
        klog.V(2).Infof("driver userAgent: %s", userAgent)

        cloud, err := azureutils.GetCloudProvider(context.Background(), kubeconfig, d.cloudConfigSecretName, d.cloudConfigSecretNamespace,
                userAgent, d.allowEmptyCloudConfig, d.enableTrafficManager, d.trafficManagerPort)
        if err != nil {
                klog.Fatalf("failed to get Azure Cloud Provider, error: %v", err)
        }
        d.cloud = cloud
        d.kubeconfig = kubeconfig

        if d.cloud != nil {
                if d.vmType != "" {
                        klog.V(2).Infof("override VMType(%s) in cloud config as %s", d.cloud.VMType, d.vmType)
                        d.cloud.VMType = d.vmType
                }

                if d.NodeID == "" {
                        // Disable UseInstanceMetadata for controller to mitigate a timeout issue using IMDS
                        // https://github.com/kubernetes-sigs/azuredisk-csi-driver/issues/168
                        klog.V(2).Infof("disable UseInstanceMetadata for controller")
                        d.cloud.Config.UseInstanceMetadata = false

                        if d.cloud.VMType == azurecloudconsts.VMTypeStandard && d.cloud.DisableAvailabilitySetNodes {
                                klog.V(2).Infof("set DisableAvailabilitySetNodes as false since VMType is %s", d.cloud.VMType)
                                d.cloud.DisableAvailabilitySetNodes = false
                        }

                        if d.cloud.VMType == azurecloudconsts.VMTypeVMSS && !d.cloud.DisableAvailabilitySetNodes && disableAVSetNodes {
                                klog.V(2).Infof("DisableAvailabilitySetNodes for controller since current VMType is vmss")
                                d.cloud.DisableAvailabilitySetNodes = true
                        }
                        klog.V(2).Infof("cloud: %s, location: %s, rg: %s, VMType: %s, PrimaryScaleSetName: %s, PrimaryAvailabilitySetName: %s, DisableAvailabilitySetNodes: %v", d.cloud.Cloud, d.cloud.Location, d.cloud.ResourceGroup, d.cloud.VMType, d.cloud.PrimaryScaleSetName, d.cloud.PrimaryAvailabilitySetName, d.cloud.DisableAvailabilitySetNodes)
                }

                if d.vmssCacheTTLInSeconds > 0 {
                        klog.V(2).Infof("reset vmssCacheTTLInSeconds as %d", d.vmssCacheTTLInSeconds)
                        d.cloud.VMCacheTTLInSeconds = int(d.vmssCacheTTLInSeconds)
                        d.cloud.VmssCacheTTLInSeconds = int(d.vmssCacheTTLInSeconds)
                }

                if d.cloud.ManagedDiskController != nil {
                        d.cloud.DisableUpdateCache = d.disableUpdateCache
                        d.cloud.AttachDetachInitialDelayInMs = int(d.attachDetachInitialDelayInMs)
                }
        }

        d.deviceHelper = optimization.NewSafeDeviceHelper()

        if d.getPerfOptimizationEnabled() {
                d.nodeInfo, err = optimization.NewNodeInfo(context.TODO(), d.getCloud(), d.NodeID)
                if err != nil {
                        klog.Warningf("Failed to get node info. Error: %v", err)
                }
        }

        d.mounter, err = mounter.NewSafeMounter(d.enableWindowsHostProcess, d.useCSIProxyGAInterface)
        if err != nil {
                klog.Fatalf("Failed to get safe mounter. Error: %v", err)
        }

        controllerCap := []csi.ControllerServiceCapability_RPC_Type{
                csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME,
                csi.ControllerServiceCapability_RPC_PUBLISH_UNPUBLISH_VOLUME,
                csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT,
                csi.ControllerServiceCapability_RPC_CLONE_VOLUME,
                csi.ControllerServiceCapability_RPC_EXPAND_VOLUME,
                csi.ControllerServiceCapability_RPC_SINGLE_NODE_MULTI_WRITER,
        }
        if d.enableListVolumes {
                controllerCap = append(controllerCap, csi.ControllerServiceCapability_RPC_LIST_VOLUMES, csi.ControllerServiceCapability_RPC_LIST_VOLUMES_PUBLISHED_NODES)
        }
        if d.enableListSnapshots {
                controllerCap = append(controllerCap, csi.ControllerServiceCapability_RPC_LIST_SNAPSHOTS)
        }

        d.AddControllerServiceCapabilities(controllerCap)
        d.AddVolumeCapabilityAccessModes(
                []csi.VolumeCapability_AccessMode_Mode{
                        csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,
                        csi.VolumeCapability_AccessMode_SINGLE_NODE_READER_ONLY,
                        csi.VolumeCapability_AccessMode_SINGLE_NODE_SINGLE_WRITER,
                        csi.VolumeCapability_AccessMode_SINGLE_NODE_MULTI_WRITER,
                })
        d.AddNodeServiceCapabilities([]csi.NodeServiceCapability_RPC_Type{
                csi.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME,
                csi.NodeServiceCapability_RPC_EXPAND_VOLUME,
                csi.NodeServiceCapability_RPC_GET_VOLUME_STATS,
                csi.NodeServiceCapability_RPC_SINGLE_NODE_MULTI_WRITER,
        })

        s := csicommon.NewNonBlockingGRPCServer()
        // Driver d act as IdentityServer, ControllerServer and NodeServer
        s.Start(endpoint, d, d, d, testingMock)
        s.Wait()
}

func (d *Driver) isGetDiskThrottled() bool {
        cache, err := d.getDiskThrottlingCache.Get(consts.ThrottlingKey, azcache.CacheReadTypeDefault)
        if err != nil {
                klog.Warningf("getDiskThrottlingCache(%s) return with error: %s", consts.ThrottlingKey, err)
                return false
        }
        return cache != nil
}

func (d *Driver) checkDiskExists(ctx context.Context, diskURI string) (*compute.Disk, error) {
        diskName, err := azureutils.GetDiskName(diskURI)
        if err != nil {
                return nil, err
        }

        resourceGroup, err := azureutils.GetResourceGroupFromURI(diskURI)
        if err != nil {
                return nil, err
        }

        if d.isGetDiskThrottled() {
                klog.Warningf("skip checkDiskExists(%s) since it's still in throttling", diskURI)
                return nil, nil
        }
        subsID := azureutils.GetSubscriptionIDFromURI(diskURI)
        disk, rerr := d.cloud.DisksClient.Get(ctx, subsID, resourceGroup, diskName)
        if rerr != nil {
                if rerr.IsThrottled() || strings.Contains(rerr.RawError.Error(), consts.RateLimited) {
                        klog.Warningf("checkDiskExists(%s) is throttled with error: %v", diskURI, rerr.Error())
                        d.getDiskThrottlingCache.Set(consts.ThrottlingKey, "")
                        return nil, nil
                }
                return nil, rerr.Error()
        }

        return &disk, nil
}

func (d *Driver) checkDiskCapacity(ctx context.Context, subsID, resourceGroup, diskName string, requestGiB int) (bool, error) {
        if d.isGetDiskThrottled() {
                klog.Warningf("skip checkDiskCapacity(%s, %s) since it's still in throttling", resourceGroup, diskName)
                return true, nil
        }

        disk, rerr := d.cloud.DisksClient.Get(ctx, subsID, resourceGroup, diskName)
        // Because we can not judge the reason of the error. Maybe the disk does not exist.
        // So here we do not handle the error.
        if rerr == nil {
                if !reflect.DeepEqual(disk, compute.Disk{}) && disk.DiskSizeGB != nil && int(*disk.DiskSizeGB) != requestGiB {
                        return false, status.Errorf(codes.AlreadyExists, "the request volume already exists, but its capacity(%v) is different from (%v)", *disk.DiskProperties.DiskSizeGB, requestGiB)
                }
        } else {
                if rerr.IsThrottled() || strings.Contains(rerr.RawError.Error(), consts.RateLimited) {
                        klog.Warningf("checkDiskCapacity(%s, %s) is throttled with error: %v", resourceGroup, diskName, rerr.Error())
                        d.getDiskThrottlingCache.Set(consts.ThrottlingKey, "")
                }
        }
        return true, nil
}

func (d *Driver) getVolumeLocks() *volumehelper.VolumeLocks {
        return d.volumeLocks
}

// setControllerCapabilities sets the controller capabilities field. It is intended for use with unit tests.
func (d *DriverCore) setControllerCapabilities(caps []*csi.ControllerServiceCapability) {
        d.Cap = caps
}

// setNodeCapabilities sets the node capabilities field. It is intended for use with unit tests.
func (d *DriverCore) setNodeCapabilities(nodeCaps []*csi.NodeServiceCapability) {
        d.NSCap = nodeCaps
}

// setName sets the Name field. It is intended for use with unit tests.
func (d *DriverCore) setName(name string) {
        d.Name = name
}

// setName sets the NodeId field. It is intended for use with unit tests.
func (d *DriverCore) setNodeID(nodeID string) {
        d.NodeID = nodeID
}

// setName sets the Version field. It is intended for use with unit tests.
func (d *DriverCore) setVersion(version string) {
        d.Version = version
}

// getCloud returns the value of the cloud field. It is intended for use with unit tests.
func (d *DriverCore) getCloud() *azure.Cloud {
        return d.cloud
}

// setCloud sets the cloud field. It is intended for use with unit tests.
func (d *DriverCore) setCloud(cloud *azure.Cloud) {
        d.cloud = cloud
}

// getMounter returns the value of the mounter field. It is intended for use with unit tests.
func (d *DriverCore) getMounter() *mount.SafeFormatAndMount {
        return d.mounter
}

// setMounter sets the mounter field. It is intended for use with unit tests.
func (d *DriverCore) setMounter(mounter *mount.SafeFormatAndMount) {
        d.mounter = mounter
}

// getPerfOptimizationEnabled returns the value of the perfOptimizationEnabled field. It is intended for use with unit tests.
func (d *DriverCore) getPerfOptimizationEnabled() bool {
        return d.perfOptimizationEnabled
}

// setPerfOptimizationEnabled sets the value of the perfOptimizationEnabled field. It is intended for use with unit tests.
func (d *DriverCore) setPerfOptimizationEnabled(enabled bool) {
        d.perfOptimizationEnabled = enabled
}

// getDeviceHelper returns the value of the deviceHelper field. It is intended for use with unit tests.
func (d *DriverCore) getDeviceHelper() optimization.Interface {
        return d.deviceHelper
}

// getNodeInfo returns the value of the nodeInfo field. It is intended for use with unit tests.
func (d *DriverCore) getNodeInfo() *optimization.NodeInfo {
        return d.nodeInfo
}

func (d *DriverCore) getHostUtil() hostUtil {
        return d.hostUtil
}

// waitForSnapshotCopy wait for copy incremental snapshot to a new region until completionPercent is 100.0
func (d *DriverCore) waitForSnapshotCopy(ctx context.Context, subsID, resourceGroup, copySnapshotName string, intervel, timeout time.Duration) error {
        timeAfter := time.After(timeout)
        timeTick := time.Tick(intervel)

        for {
                select {
                case <-timeTick:
                        copySnapshot, rerr := d.cloud.SnapshotsClient.Get(ctx, subsID, resourceGroup, copySnapshotName)
                        if rerr != nil {
                                return rerr.Error()
                        }

                        completionPercent := *copySnapshot.SnapshotProperties.CompletionPercent
                        klog.V(2).Infof("copy snapshot(%s) under rg(%s) region(%s) completionPercent: %f", copySnapshotName, resourceGroup, *copySnapshot.Location, completionPercent)
                        if completionPercent >= float64(100.0) {
                                klog.V(2).Infof("copy snapshot(%s) under rg(%s) region(%s) complete", copySnapshotName, resourceGroup, *copySnapshot.Location)
                                return nil
                        }
                case <-timeAfter:
                        return fmt.Errorf("timeout waiting for copy snapshot(%s) under rg(%s)", copySnapshotName, resourceGroup)
                }
        }
}

// getNodeInfoFromLabels get zone, instanceType from node labels
func getNodeInfoFromLabels(ctx context.Context, nodeName string, kubeClient clientset.Interface) (string, string, error) {
        if kubeClient == nil || kubeClient.CoreV1() == nil {
                return "", "", fmt.Errorf("kubeClient is nil")
        }

        node, err := kubeClient.CoreV1().Nodes().Get(ctx, nodeName, metav1.GetOptions{})
        if err != nil {
                return "", "", fmt.Errorf("get node(%s) failed with %v", nodeName, err)
        }

        if len(node.Labels) == 0 {
                return "", "", fmt.Errorf("node(%s) label is empty", nodeName)
        }
        return node.Labels[consts.WellKnownTopologyKey], node.Labels[consts.InstanceTypeKey], nil
}

// getDefaultDiskIOPSReadWrite according to requestGiB
//
//        ref: https://docs.microsoft.com/en-us/azure/virtual-machines/disks-types#ultra-disk-iops
func getDefaultDiskIOPSReadWrite(requestGiB int) int {
        iops := azurecloudconsts.DefaultDiskIOPSReadWrite
        if requestGiB > iops {
                iops = requestGiB
        }
        if iops > 160000 {
                iops = 160000
        }
        return iops
}

// getDefaultDiskMBPSReadWrite according to requestGiB
//
//        ref: https://docs.microsoft.com/en-us/azure/virtual-machines/disks-types#ultra-disk-throughput
func getDefaultDiskMBPSReadWrite(requestGiB int) int {
        bandwidth := azurecloudconsts.DefaultDiskMBpsReadWrite
        iops := getDefaultDiskIOPSReadWrite(requestGiB)
        if iops/256 > bandwidth {
                bandwidth = int(volumehelper.RoundUpSize(int64(iops), 256))
        }
        if bandwidth > iops/4 {
                bandwidth = int(volumehelper.RoundUpSize(int64(iops), 4))
        }
        if bandwidth > 4000 {
                bandwidth = 4000
        }
        return bandwidth
}

// getVMSSInstanceName get instance name from vmss compute name, e.g. "aks-agentpool-20657377-vmss_2" -> "aks-agentpool-20657377-vmss000002"
func getVMSSInstanceName(computeName string) (string, error) {
        names := strings.Split(computeName, "_")
        if len(names) != 2 {
                return "", fmt.Errorf("invalid vmss compute name: %s", computeName)
        }

        instanceID, err := strconv.Atoi(names[1])
        if err != nil {
                return "", fmt.Errorf("parsing vmss compute name(%s) failed with %v", computeName, err)
        }
        return fmt.Sprintf("%s%06s", names[0], strconv.FormatInt(int64(instanceID), 36)), nil
}

1	/*
2	Copyright 2017 The Kubernetes Authors.
3
4	Licensed under the Apache License, Version 2.0 (the "License");
5	you may not use this file except in compliance with the License.
6	You may obtain a copy of the License at
7
8	http://www.apache.org/licenses/LICENSE-2.0
9
10	Unless required by applicable law or agreed to in writing, software
11	distributed under the License is distributed on an "AS IS" BASIS,
12	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	See the License for the specific language governing permissions and
14	limitations under the License.
15	*/
16
17	package azuredisk
18
19	import (
20	"context"
21	"fmt"
22	"reflect"
23	"strconv"
24	"strings"
25	"time"
26
27	"github.com/Azure/azure-sdk-for-go/services/compute/mgmt/2022-08-01/compute"
28	"github.com/container-storage-interface/spec/lib/go/csi"
29
30	"google.golang.org/grpc/codes"
31	"google.golang.org/grpc/status"
32
33	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
34	clientset "k8s.io/client-go/kubernetes"
35	"k8s.io/klog/v2"
36	"k8s.io/kubernetes/pkg/volume/util/hostutil"
37	"k8s.io/mount-utils"
38
39	consts "sigs.k8s.io/azuredisk-csi-driver/pkg/azureconstants"
40	"sigs.k8s.io/azuredisk-csi-driver/pkg/azureutils"
41	csicommon "sigs.k8s.io/azuredisk-csi-driver/pkg/csi-common"
42	"sigs.k8s.io/azuredisk-csi-driver/pkg/mounter"
43	"sigs.k8s.io/azuredisk-csi-driver/pkg/optimization"
44	volumehelper "sigs.k8s.io/azuredisk-csi-driver/pkg/util"
45	azcache "sigs.k8s.io/cloud-provider-azure/pkg/cache"
46	azurecloudconsts "sigs.k8s.io/cloud-provider-azure/pkg/consts"
47	azure "sigs.k8s.io/cloud-provider-azure/pkg/provider"
48	)
49
50	// DriverOptions defines driver parameters specified in driver deployment
51	type DriverOptions struct {
52	NodeID string
53	DriverName string
54	VolumeAttachLimit int64
55	EnablePerfOptimization bool
56	CloudConfigSecretName string
57	CloudConfigSecretNamespace string
58	CustomUserAgent string
59	UserAgentSuffix string
60	UseCSIProxyGAInterface bool
61	EnableDiskOnlineResize bool
62	AllowEmptyCloudConfig bool
63	EnableAsyncAttach bool
64	EnableListVolumes bool
65	EnableListSnapshots bool
66	SupportZone bool
67	GetNodeInfoFromLabels bool
68	EnableDiskCapacityCheck bool
69	DisableUpdateCache bool
70	EnableTrafficManager bool
71	TrafficManagerPort int64
72	AttachDetachInitialDelayInMs int64
73	VMSSCacheTTLInSeconds int64
74	VMType string
75	EnableWindowsHostProcess bool
76	GetNodeIDFromIMDS bool
77	}
78
79	// CSIDriver defines the interface for a CSI driver.
80	type CSIDriver interface {
81	csi.ControllerServer
82	csi.NodeServer
83	csi.IdentityServer
84
85	Run(endpoint, kubeconfig string, disableAVSetNodes, testMode bool)
86	}
87
88	type hostUtil interface {
89	PathIsDevice(string) (bool, error)
90	}
91
92	// DriverCore contains fields common to both the V1 and V2 driver, and implements all interfaces of CSI drivers
93	type DriverCore struct {
94	csicommon.CSIDriver
95	perfOptimizationEnabled bool
96	cloudConfigSecretName string
97	cloudConfigSecretNamespace string
98	customUserAgent string
99	userAgentSuffix string
100	kubeconfig string
101	cloud *azure.Cloud
102	mounter *mount.SafeFormatAndMount
103	deviceHelper optimization.Interface
104	nodeInfo *optimization.NodeInfo
105	ioHandler azureutils.IOHandler
106	hostUtil hostUtil
107	useCSIProxyGAInterface bool
108	enableDiskOnlineResize bool
109	allowEmptyCloudConfig bool
110	enableAsyncAttach bool
111	enableListVolumes bool
112	enableListSnapshots bool
113	supportZone bool
114	getNodeInfoFromLabels bool
115	enableDiskCapacityCheck bool
116	disableUpdateCache bool
117	enableTrafficManager bool
118	trafficManagerPort int64
119	vmssCacheTTLInSeconds int64
120	attachDetachInitialDelayInMs int64
121	vmType string
122	enableWindowsHostProcess bool
123	getNodeIDFromIMDS bool
124	}
125
126	// Driver is the v1 implementation of the Azure Disk CSI Driver.
127	type Driver struct {
128	DriverCore
129	volumeLocks *volumehelper.VolumeLocks
130	// a timed cache GetDisk throttling
131	getDiskThrottlingCache azcache.Resource
132	}
133
134	// newDriverV1 Creates a NewCSIDriver object. Assumes vendor version is equal to driver version &
135	// does not support optional driver plugin info manifest field. Refer to CSI spec for more details.
136	func newDriverV1(options DriverOptions) Driver {	2✔
137	driver := Driver{}	2✔
138	driver.Name = options.DriverName	2✔
139	driver.Version = driverVersion	2✔
140	driver.NodeID = options.NodeID	2✔
141	driver.VolumeAttachLimit = options.VolumeAttachLimit	2✔
142	driver.perfOptimizationEnabled = options.EnablePerfOptimization	2✔
143	driver.cloudConfigSecretName = options.CloudConfigSecretName	2✔
144	driver.cloudConfigSecretNamespace = options.CloudConfigSecretNamespace	2✔
145	driver.customUserAgent = options.CustomUserAgent	2✔
146	driver.userAgentSuffix = options.UserAgentSuffix	2✔
147	driver.useCSIProxyGAInterface = options.UseCSIProxyGAInterface	2✔
148	driver.enableDiskOnlineResize = options.EnableDiskOnlineResize	2✔
149	driver.allowEmptyCloudConfig = options.AllowEmptyCloudConfig	2✔
150	driver.enableAsyncAttach = options.EnableAsyncAttach	2✔
151	driver.enableListVolumes = options.EnableListVolumes	2✔
152	driver.enableListSnapshots = options.EnableListVolumes	2✔
153	driver.supportZone = options.SupportZone	2✔
154	driver.getNodeInfoFromLabels = options.GetNodeInfoFromLabels	2✔
155	driver.enableDiskCapacityCheck = options.EnableDiskCapacityCheck	2✔
156	driver.disableUpdateCache = options.DisableUpdateCache	2✔
157	driver.attachDetachInitialDelayInMs = options.AttachDetachInitialDelayInMs	2✔
158	driver.enableTrafficManager = options.EnableTrafficManager	2✔
159	driver.trafficManagerPort = options.TrafficManagerPort	2✔
160	driver.vmssCacheTTLInSeconds = options.VMSSCacheTTLInSeconds	2✔
161	driver.vmType = options.VMType	2✔
162	driver.enableWindowsHostProcess = options.EnableWindowsHostProcess	2✔
163	driver.getNodeIDFromIMDS = options.GetNodeIDFromIMDS	2✔
164	driver.volumeLocks = volumehelper.NewVolumeLocks()	2✔
165	driver.ioHandler = azureutils.NewOSIOHandler()	2✔
166	driver.hostUtil = hostutil.NewHostUtil()	2✔
167		2✔
168	topologyKey = fmt.Sprintf("topology.%s/zone", driver.Name)	2✔
169		2✔
170	cache, err := azcache.NewTimedCache(5*time.Minute, func(key string) (interface{}, error) {	2✔
171	return nil, nil	×
172	}, false)	×
173	if err != nil {	2✔
174	klog.Fatalf("%v", err)	×
175	}	×
176	driver.getDiskThrottlingCache = cache	2✔
177	return &driver	2✔
178	}
179
180	// Run driver initialization
181	func (d *Driver) Run(endpoint, kubeconfig string, disableAVSetNodes, testingMock bool) {	4✔
182	versionMeta, err := GetVersionYAML(d.Name)	4✔
183	if err != nil {	4✔
184	klog.Fatalf("%v", err)	×
185	}	×
186	klog.Infof("\nDRIVER INFORMATION:\n-------------------\n%s\n\nStreaming logs below:", versionMeta)	4✔
187		4✔
188	userAgent := GetUserAgent(d.Name, d.customUserAgent, d.userAgentSuffix)	4✔
189	klog.V(2).Infof("driver userAgent: %s", userAgent)	4✔
190		4✔
191	cloud, err := azureutils.GetCloudProvider(context.Background(), kubeconfig, d.cloudConfigSecretName, d.cloudConfigSecretNamespace,	4✔
192	userAgent, d.allowEmptyCloudConfig, d.enableTrafficManager, d.trafficManagerPort)	4✔
193	if err != nil {	4✔
194	klog.Fatalf("failed to get Azure Cloud Provider, error: %v", err)	×
195	}	×
196	d.cloud = cloud	4✔
197	d.kubeconfig = kubeconfig	4✔
198		4✔
199	if d.cloud != nil {	8✔
200	if d.vmType != "" {	5✔
201	klog.V(2).Infof("override VMType(%s) in cloud config as %s", d.cloud.VMType, d.vmType)	1✔
202	d.cloud.VMType = d.vmType	1✔
203	}	1✔
204
205	if d.NodeID == "" {	6✔
206	// Disable UseInstanceMetadata for controller to mitigate a timeout issue using IMDS	2✔
207	// https://github.com/kubernetes-sigs/azuredisk-csi-driver/issues/168	2✔
208	klog.V(2).Infof("disable UseInstanceMetadata for controller")	2✔
209	d.cloud.Config.UseInstanceMetadata = false	2✔
210		2✔
211	if d.cloud.VMType == azurecloudconsts.VMTypeStandard && d.cloud.DisableAvailabilitySetNodes {	2✔
212	klog.V(2).Infof("set DisableAvailabilitySetNodes as false since VMType is %s", d.cloud.VMType)	×
213	d.cloud.DisableAvailabilitySetNodes = false	×
214	}	×
215
216	if d.cloud.VMType == azurecloudconsts.VMTypeVMSS && !d.cloud.DisableAvailabilitySetNodes && disableAVSetNodes {	3✔
217	klog.V(2).Infof("DisableAvailabilitySetNodes for controller since current VMType is vmss")	1✔
218	d.cloud.DisableAvailabilitySetNodes = true	1✔
219	}	1✔
220	klog.V(2).Infof("cloud: %s, location: %s, rg: %s, VMType: %s, PrimaryScaleSetName: %s, PrimaryAvailabilitySetName: %s, DisableAvailabilitySetNodes: %v", d.cloud.Cloud, d.cloud.Location, d.cloud.ResourceGroup, d.cloud.VMType, d.cloud.PrimaryScaleSetName, d.cloud.PrimaryAvailabilitySetName, d.cloud.DisableAvailabilitySetNodes)	2✔
221	}
222
223	if d.vmssCacheTTLInSeconds > 0 {	5✔
224	klog.V(2).Infof("reset vmssCacheTTLInSeconds as %d", d.vmssCacheTTLInSeconds)	1✔
225	d.cloud.VMCacheTTLInSeconds = int(d.vmssCacheTTLInSeconds)	1✔
226	d.cloud.VmssCacheTTLInSeconds = int(d.vmssCacheTTLInSeconds)	1✔
227	}	1✔
228
229	if d.cloud.ManagedDiskController != nil {	7✔
230	d.cloud.DisableUpdateCache = d.disableUpdateCache	3✔
231	d.cloud.AttachDetachInitialDelayInMs = int(d.attachDetachInitialDelayInMs)	3✔
232	}	3✔
233	}
234
235	d.deviceHelper = optimization.NewSafeDeviceHelper()	4✔
236		4✔
237	if d.getPerfOptimizationEnabled() {	5✔
238	d.nodeInfo, err = optimization.NewNodeInfo(context.TODO(), d.getCloud(), d.NodeID)	1✔
239	if err != nil {	2✔
240	klog.Warningf("Failed to get node info. Error: %v", err)	1✔
241	}	1✔
242	}
243
244	d.mounter, err = mounter.NewSafeMounter(d.enableWindowsHostProcess, d.useCSIProxyGAInterface)	4✔
245	if err != nil {	4✔
246	klog.Fatalf("Failed to get safe mounter. Error: %v", err)	×
247	}	×
248
249	controllerCap := []csi.ControllerServiceCapability_RPC_Type{	4✔
250	csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME,	4✔
251	csi.ControllerServiceCapability_RPC_PUBLISH_UNPUBLISH_VOLUME,	4✔
252	csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT,	4✔
253	csi.ControllerServiceCapability_RPC_CLONE_VOLUME,	4✔
254	csi.ControllerServiceCapability_RPC_EXPAND_VOLUME,	4✔
255	csi.ControllerServiceCapability_RPC_SINGLE_NODE_MULTI_WRITER,	4✔
256	}	4✔
257	if d.enableListVolumes {	5✔
258	controllerCap = append(controllerCap, csi.ControllerServiceCapability_RPC_LIST_VOLUMES, csi.ControllerServiceCapability_RPC_LIST_VOLUMES_PUBLISHED_NODES)	1✔
259	}	1✔
260	if d.enableListSnapshots {	5✔
261	controllerCap = append(controllerCap, csi.ControllerServiceCapability_RPC_LIST_SNAPSHOTS)	1✔
262	}	1✔
263
264	d.AddControllerServiceCapabilities(controllerCap)	4✔
265	d.AddVolumeCapabilityAccessModes(	4✔
266	[]csi.VolumeCapability_AccessMode_Mode{	4✔
267	csi.VolumeCapability_AccessMode_SINGLE_NODE_WRITER,	4✔
268	csi.VolumeCapability_AccessMode_SINGLE_NODE_READER_ONLY,	4✔
269	csi.VolumeCapability_AccessMode_SINGLE_NODE_SINGLE_WRITER,	4✔
270	csi.VolumeCapability_AccessMode_SINGLE_NODE_MULTI_WRITER,	4✔
271	})	4✔
272	d.AddNodeServiceCapabilities([]csi.NodeServiceCapability_RPC_Type{	4✔
273	csi.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME,	4✔
274	csi.NodeServiceCapability_RPC_EXPAND_VOLUME,	4✔
275	csi.NodeServiceCapability_RPC_GET_VOLUME_STATS,	4✔
276	csi.NodeServiceCapability_RPC_SINGLE_NODE_MULTI_WRITER,	4✔
277	})	4✔
278		4✔
279	s := csicommon.NewNonBlockingGRPCServer()	4✔
280	// Driver d act as IdentityServer, ControllerServer and NodeServer	4✔
281	s.Start(endpoint, d, d, d, testingMock)	4✔
282	s.Wait()	4✔
283	}
284
285	func (d *Driver) isGetDiskThrottled() bool {	13✔
286	cache, err := d.getDiskThrottlingCache.Get(consts.ThrottlingKey, azcache.CacheReadTypeDefault)	13✔
287	if err != nil {	13✔
288	klog.Warningf("getDiskThrottlingCache(%s) return with error: %s", consts.ThrottlingKey, err)	×
289	return false	×
290	}	×
291	return cache != nil	13✔
292	}
293
294	func (d Driver) checkDiskExists(ctx context.Context, diskURI string) (compute.Disk, error) {	9✔
295	diskName, err := azureutils.GetDiskName(diskURI)	9✔
296	if err != nil {	12✔
297	return nil, err	3✔
298	}	3✔
299
300	resourceGroup, err := azureutils.GetResourceGroupFromURI(diskURI)	6✔
301	if err != nil {	6✔
302	return nil, err	×
303	}	×
304
305	if d.isGetDiskThrottled() {	7✔
306	klog.Warningf("skip checkDiskExists(%s) since it's still in throttling", diskURI)	1✔
307	return nil, nil	1✔
308	}	1✔
309	subsID := azureutils.GetSubscriptionIDFromURI(diskURI)	5✔
310	disk, rerr := d.cloud.DisksClient.Get(ctx, subsID, resourceGroup, diskName)	5✔
311	if rerr != nil {	5✔
312	if rerr.IsThrottled() \|\| strings.Contains(rerr.RawError.Error(), consts.RateLimited) {	×
313	klog.Warningf("checkDiskExists(%s) is throttled with error: %v", diskURI, rerr.Error())	×
314	d.getDiskThrottlingCache.Set(consts.ThrottlingKey, "")	×
315	return nil, nil	×
316	}	×
317	return nil, rerr.Error()	×
318	}
319
320	return &disk, nil	5✔
321	}
322
323	func (d *Driver) checkDiskCapacity(ctx context.Context, subsID, resourceGroup, diskName string, requestGiB int) (bool, error) {	3✔
324	if d.isGetDiskThrottled() {	4✔
325	klog.Warningf("skip checkDiskCapacity(%s, %s) since it's still in throttling", resourceGroup, diskName)	1✔
326	return true, nil	1✔
327	}	1✔
328
329	disk, rerr := d.cloud.DisksClient.Get(ctx, subsID, resourceGroup, diskName)	2✔
330	// Because we can not judge the reason of the error. Maybe the disk does not exist.	2✔
331	// So here we do not handle the error.	2✔
332	if rerr == nil {	4✔
333	if !reflect.DeepEqual(disk, compute.Disk{}) && disk.DiskSizeGB != nil && int(*disk.DiskSizeGB) != requestGiB {	3✔
334	return false, status.Errorf(codes.AlreadyExists, "the request volume already exists, but its capacity(%v) is different from (%v)", *disk.DiskProperties.DiskSizeGB, requestGiB)	1✔
335	}	1✔
336	} else {	×
337	if rerr.IsThrottled() \|\| strings.Contains(rerr.RawError.Error(), consts.RateLimited) {	×
338	klog.Warningf("checkDiskCapacity(%s, %s) is throttled with error: %v", resourceGroup, diskName, rerr.Error())	×
339	d.getDiskThrottlingCache.Set(consts.ThrottlingKey, "")	×
340	}	×
341	}
342	return true, nil	1✔
343	}
344
345	func (d Driver) getVolumeLocks() volumehelper.VolumeLocks {	4✔
346	return d.volumeLocks	4✔
347	}	4✔
348
349	// setControllerCapabilities sets the controller capabilities field. It is intended for use with unit tests.
350	func (d DriverCore) setControllerCapabilities(caps []csi.ControllerServiceCapability) {	3✔
351	d.Cap = caps	3✔
352	}	3✔
353
354	// setNodeCapabilities sets the node capabilities field. It is intended for use with unit tests.
355	func (d DriverCore) setNodeCapabilities(nodeCaps []csi.NodeServiceCapability) {	1✔
356	d.NSCap = nodeCaps	1✔
357	}	1✔
358
359	// setName sets the Name field. It is intended for use with unit tests.
360	func (d *DriverCore) setName(name string) {	1✔
361	d.Name = name	1✔
362	}	1✔
363
364	// setName sets the NodeId field. It is intended for use with unit tests.
365	func (d *DriverCore) setNodeID(nodeID string) {	1✔
366	d.NodeID = nodeID	1✔
367	}	1✔
368
369	// setName sets the Version field. It is intended for use with unit tests.
370	func (d *DriverCore) setVersion(version string) {	1✔
371	d.Version = version	1✔
372	}	1✔
373
374	// getCloud returns the value of the cloud field. It is intended for use with unit tests.
375	func (d DriverCore) getCloud() azure.Cloud {	94✔
376	return d.cloud	94✔
377	}	94✔
378
379	// setCloud sets the cloud field. It is intended for use with unit tests.
380	func (d DriverCore) setCloud(cloud azure.Cloud) {	12✔
381	d.cloud = cloud	12✔
382	}	12✔
383
384	// getMounter returns the value of the mounter field. It is intended for use with unit tests.
385	func (d DriverCore) getMounter() mount.SafeFormatAndMount {	4✔
386	return d.mounter	4✔
387	}	4✔
388
389	// setMounter sets the mounter field. It is intended for use with unit tests.
390	func (d DriverCore) setMounter(mounter mount.SafeFormatAndMount) {	7✔
391	d.mounter = mounter	7✔
392	}	7✔
393
394	// getPerfOptimizationEnabled returns the value of the perfOptimizationEnabled field. It is intended for use with unit tests.
395	func (d *DriverCore) getPerfOptimizationEnabled() bool {	18✔
396	return d.perfOptimizationEnabled	18✔
397	}	18✔
398
399	// setPerfOptimizationEnabled sets the value of the perfOptimizationEnabled field. It is intended for use with unit tests.
400	func (d *DriverCore) setPerfOptimizationEnabled(enabled bool) {	9✔
401	d.perfOptimizationEnabled = enabled	9✔
402	}	9✔
403
404	// getDeviceHelper returns the value of the deviceHelper field. It is intended for use with unit tests.
405	func (d *DriverCore) getDeviceHelper() optimization.Interface {	115✔
406	return d.deviceHelper	115✔
407	}	115✔
408
409	// getNodeInfo returns the value of the nodeInfo field. It is intended for use with unit tests.
410	func (d DriverCore) getNodeInfo() optimization.NodeInfo {	109✔
411	return d.nodeInfo	109✔
412	}	109✔
413
414	func (d *DriverCore) getHostUtil() hostUtil {	11✔
415	return d.hostUtil	11✔
416	}	11✔
417
418	// waitForSnapshotCopy wait for copy incremental snapshot to a new region until completionPercent is 100.0
419	func (d *DriverCore) waitForSnapshotCopy(ctx context.Context, subsID, resourceGroup, copySnapshotName string, intervel, timeout time.Duration) error {	3✔
420	timeAfter := time.After(timeout)	3✔
421	timeTick := time.Tick(intervel)	3✔
422		3✔
423	for {	15✔
424	select {	12✔
425	case <-timeTick:	11✔
426	copySnapshot, rerr := d.cloud.SnapshotsClient.Get(ctx, subsID, resourceGroup, copySnapshotName)	11✔
427	if rerr != nil {	12✔
428	return rerr.Error()	1✔
429	}	1✔
430
431	completionPercent := *copySnapshot.SnapshotProperties.CompletionPercent	10✔
432	klog.V(2).Infof("copy snapshot(%s) under rg(%s) region(%s) completionPercent: %f", copySnapshotName, resourceGroup, *copySnapshot.Location, completionPercent)	10✔
433	if completionPercent >= float64(100.0) {	11✔
434	klog.V(2).Infof("copy snapshot(%s) under rg(%s) region(%s) complete", copySnapshotName, resourceGroup, *copySnapshot.Location)	1✔
435	return nil	1✔
436	}	1✔
437	case <-timeAfter:	1✔
438	return fmt.Errorf("timeout waiting for copy snapshot(%s) under rg(%s)", copySnapshotName, resourceGroup)	1✔
439	}
440	}
441	}
442
443	// getNodeInfoFromLabels get zone, instanceType from node labels
444	func getNodeInfoFromLabels(ctx context.Context, nodeName string, kubeClient clientset.Interface) (string, string, error) {	2✔
445	if kubeClient == nil \|\| kubeClient.CoreV1() == nil {	4✔
446	return "", "", fmt.Errorf("kubeClient is nil")	2✔
447	}	2✔
448
449	node, err := kubeClient.CoreV1().Nodes().Get(ctx, nodeName, metav1.GetOptions{})	×
450	if err != nil {	×
451	return "", "", fmt.Errorf("get node(%s) failed with %v", nodeName, err)	×
452	}	×
453
454	if len(node.Labels) == 0 {	×
455	return "", "", fmt.Errorf("node(%s) label is empty", nodeName)	×
456	}	×
457	return node.Labels[consts.WellKnownTopologyKey], node.Labels[consts.InstanceTypeKey], nil	×
458	}
459
460	// getDefaultDiskIOPSReadWrite according to requestGiB
461	//
462	// ref: https://docs.microsoft.com/en-us/azure/virtual-machines/disks-types#ultra-disk-iops
463	func getDefaultDiskIOPSReadWrite(requestGiB int) int {	9✔
464	iops := azurecloudconsts.DefaultDiskIOPSReadWrite	9✔
465	if requestGiB > iops {	16✔
466	iops = requestGiB	7✔
467	}	7✔
468	if iops > 160000 {	12✔
469	iops = 160000	3✔
470	}	3✔
471	return iops	9✔
472	}
473
474	// getDefaultDiskMBPSReadWrite according to requestGiB
475	//
476	// ref: https://docs.microsoft.com/en-us/azure/virtual-machines/disks-types#ultra-disk-throughput
477	func getDefaultDiskMBPSReadWrite(requestGiB int) int {	6✔
478	bandwidth := azurecloudconsts.DefaultDiskMBpsReadWrite	6✔
479	iops := getDefaultDiskIOPSReadWrite(requestGiB)	6✔
480	if iops/256 > bandwidth {	10✔
481	bandwidth = int(volumehelper.RoundUpSize(int64(iops), 256))	4✔
482	}	4✔
483	if bandwidth > iops/4 {	6✔
484	bandwidth = int(volumehelper.RoundUpSize(int64(iops), 4))	×
485	}	×
486	if bandwidth > 4000 {	6✔
487	bandwidth = 4000	×
488	}	×
489	return bandwidth	6✔
490	}
491
492	// getVMSSInstanceName get instance name from vmss compute name, e.g. "aks-agentpool-20657377-vmss_2" -> "aks-agentpool-20657377-vmss000002"
493	func getVMSSInstanceName(computeName string) (string, error) {	5✔
494	names := strings.Split(computeName, "_")	5✔
495	if len(names) != 2 {	6✔
496	return "", fmt.Errorf("invalid vmss compute name: %s", computeName)	1✔
497	}	1✔
498
499	instanceID, err := strconv.Atoi(names[1])	4✔
500	if err != nil {	5✔
501	return "", fmt.Errorf("parsing vmss compute name(%s) failed with %v", computeName, err)	1✔
502	}	1✔
503	return fmt.Sprintf("%s%06s", names[0], strconv.FormatInt(int64(instanceID), 36)), nil	3✔
504	}

kubernetes-sigs / azuredisk-csi-driver / 5408148111

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous