13390042311

Committed 18 Feb 2025 11:59AM UTC coverage: 71.503%. Remained the same

Build # 13390042311

Build Type

Pull #1842

github

Committed by

andyzhangx

Commit Message

feat: optimize azcopy perf in volume cloning scenario

Pull Request Pull Request #1842: [release-1.23] feat: optimize azcopy perf in volume cloning scenario

Run Details

2070 of 2895 relevant lines covered (71.5%)

6.49 hits per line

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

84.72

/pkg/blob/blob.go

/*
Copyright 2019 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 blob

import (
        "context"
        "errors"
        "flag"
        "fmt"
        "os"
        "strconv"
        "strings"
        "sync"
        "time"

        "github.com/Azure/azure-sdk-for-go/services/storage/mgmt/2021-09-01/storage"
        azstorage "github.com/Azure/azure-sdk-for-go/storage"
        az "github.com/Azure/go-autorest/autorest/azure"
        "github.com/container-storage-interface/spec/lib/go/csi"
        "github.com/pborman/uuid"
        "google.golang.org/grpc"
        v1 "k8s.io/api/core/v1"
        apierror "k8s.io/apimachinery/pkg/api/errors"
        metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
        "k8s.io/client-go/kubernetes"
        "k8s.io/klog/v2"
        k8sutil "k8s.io/kubernetes/pkg/volume/util"
        mount "k8s.io/mount-utils"
        utilexec "k8s.io/utils/exec"

        csicommon "sigs.k8s.io/blob-csi-driver/pkg/csi-common"
        "sigs.k8s.io/blob-csi-driver/pkg/util"
        azcache "sigs.k8s.io/cloud-provider-azure/pkg/cache"
        "sigs.k8s.io/cloud-provider-azure/pkg/provider"
        azure "sigs.k8s.io/cloud-provider-azure/pkg/provider"
)

const (
        // DefaultDriverName holds the name of the csi-driver
        DefaultDriverName              = "blob.csi.azure.com"
        blobCSIDriverName              = "blob_csi_driver"
        separator                      = "#"
        volumeIDTemplate               = "%s#%s#%s#%s#%s#%s"
        secretNameTemplate             = "azure-storage-account-%s-secret"
        serverNameField                = "server"
        storageEndpointSuffixField     = "storageendpointsuffix"
        tagsField                      = "tags"
        matchTagsField                 = "matchtags"
        protocolField                  = "protocol"
        accountNameField               = "accountname"
        accountKeyField                = "accountkey"
        storageAccountField            = "storageaccount"
        storageAccountTypeField        = "storageaccounttype"
        skuNameField                   = "skuname"
        subscriptionIDField            = "subscriptionid"
        resourceGroupField             = "resourcegroup"
        locationField                  = "location"
        secretNameField                = "secretname"
        secretNamespaceField           = "secretnamespace"
        containerNameField             = "containername"
        containerNamePrefixField       = "containernameprefix"
        storeAccountKeyField           = "storeaccountkey"
        getLatestAccountKeyField       = "getlatestaccountkey"
        isHnsEnabledField              = "ishnsenabled"
        softDeleteBlobsField           = "softdeleteblobs"
        softDeleteContainersField      = "softdeletecontainers"
        enableBlobVersioningField      = "enableblobversioning"
        getAccountKeyFromSecretField   = "getaccountkeyfromsecret"
        storageSPNClientIDField        = "azurestoragespnclientid"
        storageSPNTenantIDField        = "azurestoragespntenantid"
        storageAuthTypeField           = "azurestorageauthtype"
        storageIentityClientIDField    = "azurestorageidentityclientid"
        storageIdentityObjectIDField   = "azurestorageidentityobjectid"
        storageIdentityResourceIDField = "azurestorageidentityresourceid"
        msiEndpointField               = "msiendpoint"
        storageAADEndpointField        = "azurestorageaadendpoint"
        keyVaultURLField               = "keyvaulturl"
        keyVaultSecretNameField        = "keyvaultsecretname"
        keyVaultSecretVersionField     = "keyvaultsecretversion"
        storageAccountNameField        = "storageaccountname"
        allowBlobPublicAccessField     = "allowblobpublicaccess"
        allowSharedKeyAccessField      = "allowsharedkeyaccess"
        requireInfraEncryptionField    = "requireinfraencryption"
        ephemeralField                 = "csi.storage.k8s.io/ephemeral"
        podNamespaceField              = "csi.storage.k8s.io/pod.namespace"
        serviceAccountTokenField       = "csi.storage.k8s.io/serviceAccount.tokens"
        clientIDField                  = "clientID"
        tenantIDField                  = "tenantID"
        mountOptionsField              = "mountoptions"
        falseValue                     = "false"
        trueValue                      = "true"
        defaultSecretAccountName       = "azurestorageaccountname"
        defaultSecretAccountKey        = "azurestorageaccountkey"
        accountSasTokenField           = "azurestorageaccountsastoken"
        msiSecretField                 = "msisecret"
        storageSPNClientSecretField    = "azurestoragespnclientsecret"
        Fuse                           = "fuse"
        Fuse2                          = "fuse2"
        NFS                            = "nfs"
        AZNFS                          = "aznfs"
        NFSv3                          = "nfsv3"
        vnetResourceGroupField         = "vnetresourcegroup"
        vnetNameField                  = "vnetname"
        subnetNameField                = "subnetname"
        accessTierField                = "accesstier"
        networkEndpointTypeField       = "networkendpointtype"
        mountPermissionsField          = "mountpermissions"
        useDataPlaneAPIField           = "usedataplaneapi"

        // See https://docs.microsoft.com/en-us/rest/api/storageservices/naming-and-referencing-containers--blobs--and-metadata#container-names
        containerNameMinLength = 3
        containerNameMaxLength = 63

        accountNotProvisioned                   = "StorageAccountIsNotProvisioned"
        tooManyRequests                         = "TooManyRequests"
        clientThrottled                         = "client throttled"
        containerBeingDeletedDataplaneAPIError  = "ContainerBeingDeleted"
        containerBeingDeletedManagementAPIError = "container is being deleted"
        statusCodeNotFound                      = "StatusCode=404"
        httpCodeNotFound                        = "HTTPStatusCode: 404"

        // containerMaxSize is the max size of the blob container. See https://docs.microsoft.com/en-us/azure/storage/blobs/scalability-targets#scale-targets-for-blob-storage
        containerMaxSize = 100 * util.TiB

        subnetTemplate = "/subscriptions/%s/resourceGroups/%s/providers/Microsoft.Network/virtualNetworks/%s/subnets/%s"

        defaultNamespace = "default"

        pvcNameKey           = "csi.storage.k8s.io/pvc/name"
        pvcNamespaceKey      = "csi.storage.k8s.io/pvc/namespace"
        pvNameKey            = "csi.storage.k8s.io/pv/name"
        pvcNameMetadata      = "${pvc.metadata.name}"
        pvcNamespaceMetadata = "${pvc.metadata.namespace}"
        pvNameMetadata       = "${pv.metadata.name}"

        VolumeID = "volumeid"

        defaultStorageEndPointSuffix = "core.windows.net"
)

var (
        supportedProtocolList = []string{Fuse, Fuse2, NFS, AZNFS}
        retriableErrors       = []string{accountNotProvisioned, tooManyRequests, statusCodeNotFound, containerBeingDeletedDataplaneAPIError, containerBeingDeletedManagementAPIError, clientThrottled}

        // azcopyCloneVolumeOptions used in volume cloning between different storage account and --check-length to false because volume data may be in changing state, copy volume is not same as current source volume,
        // set --s2s-preserve-access-tier=false to avoid BlobAccessTierNotSupportedForAccountType error in azcopy
        azcopyCloneVolumeOptions = []string{"--recursive", "--check-length=false", "--s2s-preserve-access-tier=false", "--log-level=ERROR"}
)

// DriverOptions defines driver parameters specified in driver deployment
type DriverOptions struct {
        NodeID                                 string
        DriverName                             string
        BlobfuseProxyEndpoint                  string
        EnableBlobfuseProxy                    bool
        BlobfuseProxyConnTimout                int
        EnableBlobMockMount                    bool
        AllowInlineVolumeKeyAccessWithIdentity bool
        EnableGetVolumeStats                   bool
        AppendTimeStampInCacheDir              bool
        AppendMountErrorHelpLink               bool
        MountPermissions                       uint64
        EnableAznfsMount                       bool
        VolStatsCacheExpireInMinutes           int
        SasTokenExpirationMinutes              int
        WaitForAzCopyTimeoutMinutes            int
}

func (option *DriverOptions) AddFlags() {
        flag.StringVar(&option.BlobfuseProxyEndpoint, "blobfuse-proxy-endpoint", "unix://tmp/blobfuse-proxy.sock", "blobfuse-proxy endpoint")
        flag.StringVar(&option.NodeID, "nodeid", "", "node id")
        flag.StringVar(&option.DriverName, "drivername", DefaultDriverName, "name of the driver")
        flag.BoolVar(&option.EnableBlobfuseProxy, "enable-blobfuse-proxy", false, "using blobfuse proxy for mounts")
        flag.IntVar(&option.BlobfuseProxyConnTimout, "blobfuse-proxy-connect-timeout", 5, "blobfuse proxy connection timeout(seconds)")
        flag.BoolVar(&option.EnableBlobMockMount, "enable-blob-mock-mount", false, "enable mock mount(only for testing)")
        flag.BoolVar(&option.EnableGetVolumeStats, "enable-get-volume-stats", false, "allow GET_VOLUME_STATS on agent node")
        flag.BoolVar(&option.AppendTimeStampInCacheDir, "append-timestamp-cache-dir", false, "append timestamp into cache directory on agent node")
        flag.Uint64Var(&option.MountPermissions, "mount-permissions", 0777, "mounted folder permissions")
        flag.BoolVar(&option.AllowInlineVolumeKeyAccessWithIdentity, "allow-inline-volume-key-access-with-idenitity", false, "allow accessing storage account key using cluster identity for inline volume")
        flag.BoolVar(&option.AppendMountErrorHelpLink, "append-mount-error-help-link", true, "Whether to include a link for help with mount errors when a mount error occurs.")
        flag.BoolVar(&option.EnableAznfsMount, "enable-aznfs-mount", false, "replace nfs mount with aznfs mount")
        flag.IntVar(&option.VolStatsCacheExpireInMinutes, "vol-stats-cache-expire-in-minutes", 10, "The cache expire time in minutes for volume stats cache")
        flag.IntVar(&option.SasTokenExpirationMinutes, "sas-token-expiration-minutes", 1440, "sas token expiration minutes during volume cloning")
        flag.IntVar(&option.WaitForAzCopyTimeoutMinutes, "wait-for-azcopy-timeout-minutes", 18, "timeout in minutes for waiting for azcopy to finish")
}

// Driver implements all interfaces of CSI drivers
type Driver struct {
        csicommon.CSIDriver

        cloud                 *azure.Cloud
        KubeClient            kubernetes.Interface
        blobfuseProxyEndpoint string
        // enableBlobMockMount is only for testing, DO NOT set as true in non-testing scenario
        enableBlobMockMount                    bool
        enableBlobfuseProxy                    bool
        enableGetVolumeStats                   bool
        allowInlineVolumeKeyAccessWithIdentity bool
        appendTimeStampInCacheDir              bool
        appendMountErrorHelpLink               bool
        blobfuseProxyConnTimout                int
        mountPermissions                       uint64
        enableAznfsMount                       bool
        mounter                                *mount.SafeFormatAndMount
        volLockMap                             *util.LockMap
        // A map storing all volumes with ongoing operations so that additional operations
        // for that same volume (as defined by VolumeID) return an Aborted error
        volumeLocks *volumeLocks
        // only for nfs feature
        subnetLockMap *util.LockMap
        // a map storing all volumes created by this driver <volumeName, accountName>
        volMap sync.Map
        // a timed cache storing all volumeIDs and storage accounts that are using data plane API
        dataPlaneAPIVolCache azcache.Resource
        // a timed cache storing account search history (solve account list throttling issue)
        accountSearchCache azcache.Resource
        // a timed cache storing volume stats <volumeID, volumeStats>
        volStatsCache azcache.Resource
        // a timed cache storing account which should use sastoken for azcopy based volume cloning
        azcopySasTokenCache azcache.Resource
        // sas expiry time for azcopy in volume clone
        sasTokenExpirationMinutes int
        // timeout in minutes for waiting for azcopy to finish
        waitForAzCopyTimeoutMinutes int
        // azcopy for provide exec mock for ut
        azcopy *util.Azcopy
}

// NewDriver 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 NewDriver(options *DriverOptions, kubeClient kubernetes.Interface, cloud *provider.Cloud) *Driver {
        var err error
        d := Driver{
                volLockMap:                             util.NewLockMap(),
                subnetLockMap:                          util.NewLockMap(),
                volumeLocks:                            newVolumeLocks(),
                blobfuseProxyEndpoint:                  options.BlobfuseProxyEndpoint,
                enableBlobfuseProxy:                    options.EnableBlobfuseProxy,
                allowInlineVolumeKeyAccessWithIdentity: options.AllowInlineVolumeKeyAccessWithIdentity,
                blobfuseProxyConnTimout:                options.BlobfuseProxyConnTimout,
                enableBlobMockMount:                    options.EnableBlobMockMount,
                enableGetVolumeStats:                   options.EnableGetVolumeStats,
                appendMountErrorHelpLink:               options.AppendMountErrorHelpLink,
                mountPermissions:                       options.MountPermissions,
                enableAznfsMount:                       options.EnableAznfsMount,
                sasTokenExpirationMinutes:              options.SasTokenExpirationMinutes,
                waitForAzCopyTimeoutMinutes:            options.WaitForAzCopyTimeoutMinutes,
                azcopy:                                 &util.Azcopy{},
                KubeClient:                             kubeClient,
                cloud:                                  cloud,
        }
        d.Name = options.DriverName
        d.Version = driverVersion
        d.NodeID = options.NodeID

        getter := func(_ string) (interface{}, error) { return nil, nil }
        if d.accountSearchCache, err = azcache.NewTimedCache(time.Minute, getter, false); err != nil {
                klog.Fatalf("%v", err)
        }
        if d.dataPlaneAPIVolCache, err = azcache.NewTimedCache(24*30*time.Hour, getter, false); err != nil {
                klog.Fatalf("%v", err)
        }
        if d.azcopySasTokenCache, err = azcache.NewTimedCache(15*time.Minute, getter, false); err != nil {
                klog.Fatalf("%v", err)
        }

        if options.VolStatsCacheExpireInMinutes <= 0 {
                options.VolStatsCacheExpireInMinutes = 10 // default expire in 10 minutes
        }
        if d.volStatsCache, err = azcache.NewTimedCache(time.Duration(options.VolStatsCacheExpireInMinutes)*time.Minute, getter, false); err != nil {
                klog.Fatalf("%v", err)
        }
        d.mounter = &mount.SafeFormatAndMount{
                Interface: mount.New(""),
                Exec:      utilexec.New(),
        }

        // Initialize default library driver
        d.AddControllerServiceCapabilities(
                []csi.ControllerServiceCapability_RPC_Type{
                        csi.ControllerServiceCapability_RPC_CREATE_DELETE_VOLUME,
                        //csi.ControllerServiceCapability_RPC_CREATE_DELETE_SNAPSHOT,
                        //csi.ControllerServiceCapability_RPC_LIST_SNAPSHOTS,
                        csi.ControllerServiceCapability_RPC_EXPAND_VOLUME,
                        csi.ControllerServiceCapability_RPC_SINGLE_NODE_MULTI_WRITER,
                        csi.ControllerServiceCapability_RPC_CLONE_VOLUME,
                })
        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,
                csi.VolumeCapability_AccessMode_MULTI_NODE_READER_ONLY,
                csi.VolumeCapability_AccessMode_MULTI_NODE_SINGLE_WRITER,
                csi.VolumeCapability_AccessMode_MULTI_NODE_MULTI_WRITER,
        })

        nodeCap := []csi.NodeServiceCapability_RPC_Type{
                csi.NodeServiceCapability_RPC_STAGE_UNSTAGE_VOLUME,
                csi.NodeServiceCapability_RPC_SINGLE_NODE_MULTI_WRITER,
        }
        if d.enableGetVolumeStats {
                nodeCap = append(nodeCap, csi.NodeServiceCapability_RPC_GET_VOLUME_STATS)
        }
        d.AddNodeServiceCapabilities(nodeCap)

        return &d
}

// Run driver initialization
func (d *Driver) Run(ctx context.Context, endpoint string) error {
        versionMeta, err := GetVersionYAML(d.Name)
        if err != nil {
                klog.Fatalf("%v", err)
        }
        klog.Infof("\nDRIVER INFORMATION:\n-------------------\n%s\n\nStreaming logs below:", versionMeta)
        grpcInterceptor := grpc.UnaryInterceptor(csicommon.LogGRPC)
        opts := []grpc.ServerOption{
                grpcInterceptor,
        }
        s := grpc.NewServer(opts...)
        csi.RegisterIdentityServer(s, d)
        csi.RegisterControllerServer(s, d)
        csi.RegisterNodeServer(s, d)

        go func() {
                //graceful shutdown
                <-ctx.Done()
                s.GracefulStop()
        }()
        // Driver d act as IdentityServer, ControllerServer and NodeServer
        listener, err := csicommon.Listen(ctx, endpoint)
        if err != nil {
                klog.Fatalf("failed to listen to endpoint, error: %v", err)
        }
        err = s.Serve(listener)
        if errors.Is(err, grpc.ErrServerStopped) {
                klog.Infof("gRPC server stopped serving")
                return nil
        }
        return err
}

// GetContainerInfo get container info according to volume id
// the format of VolumeId is: rg#accountName#containerName#uuid#secretNamespace#subsID
//
// e.g.
// input: "rg#f5713de20cde511e8ba4900#containerName#uuid#"
// output: rg, f5713de20cde511e8ba4900, containerName, "" , ""
// input: "rg#f5713de20cde511e8ba4900#containerName#uuid#namespace#"
// output: rg, f5713de20cde511e8ba4900, containerName, namespace, ""
// input: "rg#f5713de20cde511e8ba4900#containerName#uuid#namespace#subsID"
// output: rg, f5713de20cde511e8ba4900, containerName, namespace, subsID
func GetContainerInfo(id string) (string, string, string, string, string, error) {
        segments := strings.Split(id, separator)
        if len(segments) < 3 {
                return "", "", "", "", "", fmt.Errorf("error parsing volume id: %q, should at least contain two #", id)
        }
        var secretNamespace, subsID string
        if len(segments) > 4 {
                secretNamespace = segments[4]
        }
        if len(segments) > 5 {
                subsID = segments[5]
        }
        return segments[0], segments[1], segments[2], secretNamespace, subsID, nil
}

// A container name must be a valid DNS name, conforming to the following naming rules:
//  1. Container names must start with a letter or number, and can contain only letters, numbers, and the dash (-) character.
//  2. Every dash (-) character must be immediately preceded and followed by a letter or number; consecutive dashes are not permitted in container names.
//  3. All letters in a container name must be lowercase.
//  4. Container names must be from 3 through 63 characters long.
//
// See https://docs.microsoft.com/en-us/rest/api/storageservices/naming-and-referencing-containers--blobs--and-metadata#container-names
func getValidContainerName(volumeName, protocol string) string {
        containerName := strings.ToLower(volumeName)
        if len(containerName) > containerNameMaxLength {
                containerName = containerName[0:containerNameMaxLength]
        }
        if !checkContainerNameBeginAndEnd(containerName) || len(containerName) < containerNameMinLength {
                // now we set as 63 for maximum container name length
                // todo: get cluster name
                containerName = k8sutil.GenerateVolumeName(fmt.Sprintf("pvc-%s", protocol), uuid.NewUUID().String(), 63)
                klog.Warningf("requested volume name (%s) is invalid, regenerated as (%q)", volumeName, containerName)
        }
        return strings.Replace(containerName, "--", "-", -1)
}

func checkContainerNameBeginAndEnd(containerName string) bool {
        length := len(containerName)
        if (('a' <= containerName[0] && containerName[0] <= 'z') ||
                ('0' <= containerName[0] && containerName[0] <= '9')) &&
                (('a' <= containerName[length-1] && containerName[length-1] <= 'z') ||
                        ('0' <= containerName[length-1] && containerName[length-1] <= '9')) {
                return true
        }

        return false
}

// isSASToken checks if the key contains the patterns.
// SAS token format could refer to https://docs.microsoft.com/en-us/rest/api/eventhub/generate-sas-token
func isSASToken(key string) bool {
        return strings.HasPrefix(key, "?")
}

// GetAuthEnv return <accountName, containerName, authEnv, error>
func (d *Driver) GetAuthEnv(ctx context.Context, volumeID, protocol string, attrib, secrets map[string]string) (string, string, string, string, []string, error) {
        rgName, accountName, containerName, secretNamespace, _, err := GetContainerInfo(volumeID)
        if err != nil {
                // ignore volumeID parsing error
                klog.V(2).Infof("parsing volumeID(%s) return with error: %v", volumeID, err)
                err = nil
        }

        var (
                subsID                  string
                accountKey              string
                accountSasToken         string
                msiSecret               string
                storageSPNClientSecret  string
                storageSPNClientID      string
                storageSPNTenantID      string
                secretName              string
                pvcNamespace            string
                keyVaultURL             string
                keyVaultSecretName      string
                keyVaultSecretVersion   string
                azureStorageAuthType    string
                authEnv                 []string
                getAccountKeyFromSecret bool
                getLatestAccountKey     bool
                clientID                string
                tenantID                string
                serviceAccountToken     string
        )

        for k, v := range attrib {
                switch strings.ToLower(k) {
                case subscriptionIDField:
                        subsID = v
                case resourceGroupField:
                        rgName = v
                case containerNameField:
                        containerName = v
                case keyVaultURLField:
                        keyVaultURL = v
                case keyVaultSecretNameField:
                        keyVaultSecretName = v
                case keyVaultSecretVersionField:
                        keyVaultSecretVersion = v
                case storageAccountField:
                        accountName = v
                case storageAccountNameField: // for compatibility
                        accountName = v
                case secretNameField:
                        secretName = v
                case secretNamespaceField:
                        secretNamespace = v
                case pvcNamespaceKey:
                        pvcNamespace = v
                case getAccountKeyFromSecretField:
                        getAccountKeyFromSecret = strings.EqualFold(v, trueValue)
                case storageAuthTypeField:
                        azureStorageAuthType = v
                        authEnv = append(authEnv, "AZURE_STORAGE_AUTH_TYPE="+v)
                case storageIentityClientIDField:
                        authEnv = append(authEnv, "AZURE_STORAGE_IDENTITY_CLIENT_ID="+v)
                case storageIdentityObjectIDField:
                        authEnv = append(authEnv, "AZURE_STORAGE_IDENTITY_OBJECT_ID="+v)
                case storageIdentityResourceIDField:
                        authEnv = append(authEnv, "AZURE_STORAGE_IDENTITY_RESOURCE_ID="+v)
                case msiEndpointField:
                        authEnv = append(authEnv, "MSI_ENDPOINT="+v)
                case storageSPNClientIDField:
                        storageSPNClientID = v
                case storageSPNTenantIDField:
                        storageSPNTenantID = v
                case storageAADEndpointField:
                        authEnv = append(authEnv, "AZURE_STORAGE_AAD_ENDPOINT="+v)
                case getLatestAccountKeyField:
                        if getLatestAccountKey, err = strconv.ParseBool(v); err != nil {
                                return rgName, accountName, accountKey, containerName, authEnv, fmt.Errorf("invalid %s: %s in volume context", getLatestAccountKeyField, v)
                        }
                case strings.ToLower(clientIDField):
                        clientID = v
                case strings.ToLower(tenantIDField):
                        tenantID = v
                case strings.ToLower(serviceAccountTokenField):
                        serviceAccountToken = v
                }
        }
        klog.V(2).Infof("volumeID(%s) authEnv: %s", volumeID, authEnv)

        if protocol == NFS {
                // nfs protocol does not need account key, return directly
                return rgName, accountName, accountKey, containerName, authEnv, err
        }

        if secretNamespace == "" {
                if pvcNamespace == "" {
                        secretNamespace = defaultNamespace
                } else {
                        secretNamespace = pvcNamespace
                }
        }

        if rgName == "" {
                rgName = d.cloud.ResourceGroup
        }

        if tenantID == "" {
                tenantID = d.cloud.TenantID
        }

        // if client id is specified, we only use service account token to get account key
        if clientID != "" {
                klog.V(2).Infof("clientID(%s) is specified, use service account token to get account key", clientID)
                if subsID == "" {
                        subsID = d.cloud.SubscriptionID
                }
                accountKey, err := d.cloud.GetStorageAccesskeyFromServiceAccountToken(ctx, subsID, accountName, rgName, clientID, tenantID, serviceAccountToken)
                authEnv = append(authEnv, "AZURE_STORAGE_ACCESS_KEY="+accountKey)
                return rgName, accountName, accountKey, containerName, authEnv, err
        }

        // 1. If keyVaultURL is not nil, preferentially use the key stored in key vault.
        // 2. Then if secrets map is not nil, use the key stored in the secrets map.
        // 3. Finally if both keyVaultURL and secrets map are nil, get the key from Azure.
        if keyVaultURL != "" {
                key, err := d.getKeyVaultSecretContent(ctx, keyVaultURL, keyVaultSecretName, keyVaultSecretVersion)
                if err != nil {
                        return rgName, accountName, accountKey, containerName, authEnv, err
                }
                if isSASToken(key) {
                        accountSasToken = key
                } else {
                        accountKey = key
                }
        } else {
                if len(secrets) == 0 {
                        if secretName == "" && accountName != "" {
                                secretName = fmt.Sprintf(secretNameTemplate, accountName)
                        }
                        if secretName != "" {
                                // read from k8s secret first
                                var name, spnClientID, spnTenantID string
                                name, accountKey, accountSasToken, msiSecret, storageSPNClientSecret, spnClientID, spnTenantID, err = d.GetInfoFromSecret(ctx, secretName, secretNamespace)
                                if name != "" {
                                        accountName = name
                                }
                                if spnClientID != "" {
                                        storageSPNClientID = spnClientID
                                }
                                if spnTenantID != "" {
                                        storageSPNTenantID = spnTenantID
                                }
                                if err != nil && strings.EqualFold(azureStorageAuthType, "msi") {
                                        klog.V(2).Infof("ignore error(%v) since secret is optional for auth type(%s)", err, azureStorageAuthType)
                                        err = nil
                                }
                                if err != nil && !getAccountKeyFromSecret && (azureStorageAuthType == "" || strings.EqualFold(azureStorageAuthType, "key")) {
                                        klog.V(2).Infof("get account(%s) key from secret(%s, %s) failed with error: %v, use cluster identity to get account key instead",
                                                accountName, secretNamespace, secretName, err)
                                        accountKey, err = d.cloud.GetStorageAccesskey(ctx, subsID, accountName, rgName, getLatestAccountKey)
                                        if err != nil {
                                                return rgName, accountName, accountKey, containerName, authEnv, fmt.Errorf("no key for storage account(%s) under resource group(%s), err %w", accountName, rgName, err)
                                        }
                                }
                        }
                } else {
                        for k, v := range secrets {
                                v = strings.TrimSpace(v)
                                switch strings.ToLower(k) {
                                case accountNameField:
                                        accountName = v
                                case defaultSecretAccountName: // for compatibility with built-in blobfuse plugin
                                        accountName = v
                                case accountKeyField:
                                        accountKey = v
                                case defaultSecretAccountKey: // for compatibility with built-in blobfuse plugin
                                        accountKey = v
                                case accountSasTokenField:
                                        accountSasToken = v
                                case msiSecretField:
                                        msiSecret = v
                                case storageSPNClientSecretField:
                                        storageSPNClientSecret = v
                                case storageSPNClientIDField:
                                        storageSPNClientID = v
                                case storageSPNTenantIDField:
                                        storageSPNTenantID = v
                                }
                        }
                }
        }

        if containerName == "" {
                err = fmt.Errorf("could not find containerName from attributes(%v) or volumeID(%v)", attrib, volumeID)
        }

        if accountKey != "" {
                authEnv = append(authEnv, "AZURE_STORAGE_ACCESS_KEY="+accountKey)
        }

        if accountSasToken != "" {
                klog.V(2).Infof("accountSasToken is not empty, use it to access storage account(%s), container(%s)", accountName, containerName)
                authEnv = append(authEnv, "AZURE_STORAGE_SAS_TOKEN="+accountSasToken)
        }

        if msiSecret != "" {
                klog.V(2).Infof("msiSecret is not empty, use it to access storage account(%s), container(%s)", accountName, containerName)
                authEnv = append(authEnv, "MSI_SECRET="+msiSecret)
        }

        if storageSPNClientSecret != "" {
                klog.V(2).Infof("storageSPNClientSecret is not empty, use it to access storage account(%s), container(%s)", accountName, containerName)
                authEnv = append(authEnv, "AZURE_STORAGE_SPN_CLIENT_SECRET="+storageSPNClientSecret)
        }

        if storageSPNClientID != "" {
                klog.V(2).Infof("storageSPNClientID(%s) is not empty, use it to access storage account(%s), container(%s)", storageSPNClientID, accountName, containerName)
                authEnv = append(authEnv, "AZURE_STORAGE_SPN_CLIENT_ID="+storageSPNClientID)
        }

        if storageSPNTenantID != "" {
                klog.V(2).Infof("storageSPNTenantID(%s) is not empty, use it to access storage account(%s), container(%s)", storageSPNTenantID, accountName, containerName)
                authEnv = append(authEnv, "AZURE_STORAGE_SPN_TENANT_ID="+storageSPNTenantID)
        }

        return rgName, accountName, accountKey, containerName, authEnv, err
}

// GetStorageAccountAndContainer get storage account and container info
// returns <accountName, accountKey, accountSasToken, containerName>
// only for e2e testing
func (d *Driver) GetStorageAccountAndContainer(ctx context.Context, volumeID string, attrib, secrets map[string]string) (string, string, string, string, error) {
        var (
                subsID                string
                accountName           string
                accountKey            string
                accountSasToken       string
                containerName         string
                keyVaultURL           string
                keyVaultSecretName    string
                keyVaultSecretVersion string
                getLatestAccountKey   bool
                err                   error
        )

        for k, v := range attrib {
                switch strings.ToLower(k) {
                case subscriptionIDField:
                        subsID = v
                case containerNameField:
                        containerName = v
                case keyVaultURLField:
                        keyVaultURL = v
                case keyVaultSecretNameField:
                        keyVaultSecretName = v
                case keyVaultSecretVersionField:
                        keyVaultSecretVersion = v
                case storageAccountField:
                        accountName = v
                case storageAccountNameField: // for compatibility
                        accountName = v
                case getLatestAccountKeyField:
                        if getLatestAccountKey, err = strconv.ParseBool(v); err != nil {
                                return "", "", "", "", fmt.Errorf("invalid %s: %s in volume context", getLatestAccountKeyField, v)
                        }
                }
        }

        // 1. If keyVaultURL is not nil, preferentially use the key stored in key vault.
        // 2. Then if secrets map is not nil, use the key stored in the secrets map.
        // 3. Finally if both keyVaultURL and secrets map are nil, get the key from Azure.
        if keyVaultURL != "" {
                key, err := d.getKeyVaultSecretContent(ctx, keyVaultURL, keyVaultSecretName, keyVaultSecretVersion)
                if err != nil {
                        return "", "", "", "", err
                }
                if isSASToken(key) {
                        accountSasToken = key
                } else {
                        accountKey = key
                }
        } else {
                if len(secrets) == 0 {
                        var rgName string
                        rgName, accountName, containerName, _, _, err = GetContainerInfo(volumeID)
                        if err != nil {
                                return "", "", "", "", err
                        }

                        if rgName == "" {
                                rgName = d.cloud.ResourceGroup
                        }

                        accountKey, err = d.cloud.GetStorageAccesskey(ctx, subsID, accountName, rgName, getLatestAccountKey)
                        if err != nil {
                                return "", "", "", "", fmt.Errorf("no key for storage account(%s) under resource group(%s), err %w", accountName, rgName, err)
                        }
                }
        }

        if containerName == "" {
                return "", "", "", "", fmt.Errorf("could not find containerName from attributes(%v) or volumeID(%v)", attrib, volumeID)
        }

        return accountName, accountKey, accountSasToken, containerName, nil
}

func IsCorruptedDir(dir string) bool {
        _, pathErr := mount.PathExists(dir)
        return pathErr != nil && mount.IsCorruptedMnt(pathErr)
}

func isRetriableError(err error) bool {
        if err != nil {
                for _, v := range retriableErrors {
                        if strings.Contains(strings.ToLower(err.Error()), strings.ToLower(v)) {
                                return true
                        }
                }
        }
        return false
}

func isSupportedProtocol(protocol string) bool {
        if protocol == "" {
                return true
        }
        for _, v := range supportedProtocolList {
                if protocol == v || protocol == NFSv3 {
                        return true
                }
        }
        return false
}

func isSupportedAccessTier(accessTier string) bool {
        if accessTier == "" {
                return true
        }
        for _, tier := range storage.PossibleAccessTierValues() {
                if accessTier == string(tier) {
                        return true
                }
        }
        return false
}

// container names can contain only lowercase letters, numbers, and hyphens,
// and must begin and end with a letter or a number
func isSupportedContainerNamePrefix(prefix string) bool {
        if prefix == "" {
                return true
        }
        if len(prefix) > 20 {
                return false
        }
        if prefix[0] == '-' {
                return false
        }
        for _, v := range prefix {
                if v != '-' && (v < '0' || v > '9') && (v < 'a' || v > 'z') {
                        return false
                }
        }
        return true
}

// isNFSProtocol checks if the protocol is NFS or AZNFS
func isNFSProtocol(protocol string) bool {
        protocol = strings.ToLower(protocol)
        return protocol == NFS || protocol == AZNFS || protocol == NFSv3
}

// get storage account from secrets map
func getStorageAccount(secrets map[string]string) (string, string, error) {
        if secrets == nil {
                return "", "", fmt.Errorf("unexpected: getStorageAccount secrets is nil")
        }

        var accountName, accountKey string
        for k, v := range secrets {
                v = strings.TrimSpace(v)
                switch strings.ToLower(k) {
                case accountNameField:
                        accountName = v
                case defaultSecretAccountName: // for compatibility with built-in azurefile plugin
                        accountName = v
                case accountKeyField:
                        accountKey = v
                case defaultSecretAccountKey: // for compatibility with built-in azurefile plugin
                        accountKey = v
                }
        }

        if accountName == "" {
                return accountName, accountKey, fmt.Errorf("could not find %s or %s field in secrets", accountNameField, defaultSecretAccountName)
        }
        if accountKey == "" {
                return accountName, accountKey, fmt.Errorf("could not find %s or %s field in secrets", accountKeyField, defaultSecretAccountKey)
        }

        accountName = strings.TrimSpace(accountName)
        klog.V(4).Infof("got storage account(%s) from secret", accountName)
        return accountName, accountKey, nil
}

func getContainerReference(containerName string, secrets map[string]string, env az.Environment) (*azstorage.Container, error) {
        accountName, accountKey, rerr := getStorageAccount(secrets)
        if rerr != nil {
                return nil, rerr
        }
        client, err := azstorage.NewBasicClientOnSovereignCloud(accountName, accountKey, env)
        if err != nil {
                return nil, err
        }
        blobClient := client.GetBlobService()
        container := blobClient.GetContainerReference(containerName)
        if container == nil {
                return nil, fmt.Errorf("ContainerReference of %s is nil", containerName)
        }
        return container, nil
}

func setAzureCredentials(ctx context.Context, kubeClient kubernetes.Interface, accountName, accountKey, secretNamespace string) (string, error) {
        if kubeClient == nil {
                klog.Warningf("could not create secret: kubeClient is nil")
                return "", nil
        }
        if accountName == "" || accountKey == "" {
                return "", fmt.Errorf("the account info is not enough, accountName(%v), accountKey(%v)", accountName, accountKey)
        }
        secretName := fmt.Sprintf(secretNameTemplate, accountName)
        secret := &v1.Secret{
                ObjectMeta: metav1.ObjectMeta{
                        Namespace: secretNamespace,
                        Name:      secretName,
                },
                Data: map[string][]byte{
                        defaultSecretAccountName: []byte(accountName),
                        defaultSecretAccountKey:  []byte(accountKey),
                },
                Type: "Opaque",
        }
        _, err := kubeClient.CoreV1().Secrets(secretNamespace).Create(ctx, secret, metav1.CreateOptions{})
        if apierror.IsAlreadyExists(err) {
                err = nil
        }
        if err != nil {
                return "", fmt.Errorf("couldn't create secret %w", err)
        }
        return secretName, err
}

// GetStorageAccesskey get Azure storage account key from
//  1. secrets (if not empty)
//  2. use k8s client identity to read from k8s secret
//  3. use cluster identity to get from storage account directly
func (d *Driver) GetStorageAccesskey(ctx context.Context, accountOptions *azure.AccountOptions, secrets map[string]string, secretName, secretNamespace string) (string, string, error) {
        if len(secrets) > 0 {
                return getStorageAccount(secrets)
        }

        // read from k8s secret first
        if secretName == "" {
                secretName = fmt.Sprintf(secretNameTemplate, accountOptions.Name)
        }
        _, accountKey, _, _, _, _, _, err := d.GetInfoFromSecret(ctx, secretName, secretNamespace) //nolint
        if err != nil {
                klog.V(2).Infof("could not get account(%s) key from secret(%s) namespace(%s), error: %v, use cluster identity to get account key instead", accountOptions.Name, secretName, secretNamespace, err)
                accountKey, err = d.cloud.GetStorageAccesskey(ctx, accountOptions.SubscriptionID, accountOptions.Name, accountOptions.ResourceGroup, accountOptions.GetLatestAccountKey)
        }
        return accountOptions.Name, accountKey, err
}

// GetInfoFromSecret get info from k8s secret
// return <accountName, accountKey, accountSasToken, msiSecret, spnClientSecret, spnClientID, spnTenantID, error>
func (d *Driver) GetInfoFromSecret(ctx context.Context, secretName, secretNamespace string) (string, string, string, string, string, string, string, error) {
        if d.cloud.KubeClient == nil {
                return "", "", "", "", "", "", "", fmt.Errorf("could not get account key from secret(%s): KubeClient is nil", secretName)
        }

        secret, err := d.cloud.KubeClient.CoreV1().Secrets(secretNamespace).Get(ctx, secretName, metav1.GetOptions{})
        if err != nil {
                return "", "", "", "", "", "", "", fmt.Errorf("could not get secret(%v): %w", secretName, err)
        }

        accountName := strings.TrimSpace(string(secret.Data[defaultSecretAccountName][:]))
        accountKey := strings.TrimSpace(string(secret.Data[defaultSecretAccountKey][:]))
        accountSasToken := strings.TrimSpace(string(secret.Data[accountSasTokenField][:]))
        msiSecret := strings.TrimSpace(string(secret.Data[msiSecretField][:]))
        spnClientSecret := strings.TrimSpace(string(secret.Data[storageSPNClientSecretField][:]))
        spnClientID := strings.TrimSpace(string(secret.Data[storageSPNClientIDField][:]))
        spnTenantID := strings.TrimSpace(string(secret.Data[storageSPNTenantIDField][:]))

        klog.V(4).Infof("got storage account(%s) from secret(%s) namespace(%s)", accountName, secretName, secretNamespace)
        return accountName, accountKey, accountSasToken, msiSecret, spnClientSecret, spnClientID, spnTenantID, nil
}

// getSubnetResourceID get default subnet resource ID from cloud provider config
func (d *Driver) getSubnetResourceID(vnetResourceGroup, vnetName, subnetName string) string {
        subsID := d.cloud.SubscriptionID
        if len(d.cloud.NetworkResourceSubscriptionID) > 0 {
                subsID = d.cloud.NetworkResourceSubscriptionID
        }

        if len(vnetResourceGroup) == 0 {
                vnetResourceGroup = d.cloud.ResourceGroup
                if len(d.cloud.VnetResourceGroup) > 0 {
                        vnetResourceGroup = d.cloud.VnetResourceGroup
                }
        }

        if len(vnetName) == 0 {
                vnetName = d.cloud.VnetName
        }

        if len(subnetName) == 0 {
                subnetName = d.cloud.SubnetName
        }
        return fmt.Sprintf(subnetTemplate, subsID, vnetResourceGroup, vnetName, subnetName)
}

func (d *Driver) useDataPlaneAPI(volumeID, accountName string) bool {
        cache, err := d.dataPlaneAPIVolCache.Get(volumeID, azcache.CacheReadTypeDefault)
        if err != nil {
                klog.Errorf("get(%s) from dataPlaneAPIVolCache failed with error: %v", volumeID, err)
        }
        if cache != nil {
                return true
        }
        cache, err = d.dataPlaneAPIVolCache.Get(accountName, azcache.CacheReadTypeDefault)
        if err != nil {
                klog.Errorf("get(%s) from dataPlaneAPIVolCache failed with error: %v", accountName, err)
        }
        if cache != nil {
                return true
        }
        return false
}

// appendDefaultMountOptions return mount options combined with mountOptions and defaultMountOptions
func appendDefaultMountOptions(mountOptions []string, tmpPath, containerName string) []string {
        var defaultMountOptions = map[string]string{
                "--pre-mount-validate": "true",
                "--use-https":          "true",
                "--tmp-path":           tmpPath,
                "--container-name":     containerName,
                // prevent billing charges on mounting
                "--cancel-list-on-mount-seconds": "10",
                // allow remounting using a non-empty tmp-path
                "--empty-dir-check": "false",
        }

        // stores the mount options already included in mountOptions
        included := make(map[string]bool)

        for _, mountOption := range mountOptions {
                for k := range defaultMountOptions {
                        if strings.HasPrefix(mountOption, k) {
                                included[k] = true
                        }
                }
        }

        allMountOptions := mountOptions

        for k, v := range defaultMountOptions {
                if _, isIncluded := included[k]; !isIncluded {
                        if v != "" {
                                allMountOptions = append(allMountOptions, fmt.Sprintf("%s=%s", k, v))
                        } else {
                                allMountOptions = append(allMountOptions, k)
                        }
                }
        }

        return allMountOptions
}

// chmodIfPermissionMismatch only perform chmod when permission mismatches
func chmodIfPermissionMismatch(targetPath string, mode os.FileMode) error {
        info, err := os.Lstat(targetPath)
        if err != nil {
                return err
        }
        perm := info.Mode() & os.ModePerm
        if perm != mode {
                klog.V(2).Infof("chmod targetPath(%s, mode:0%o) with permissions(0%o)", targetPath, info.Mode(), mode)
                if err := os.Chmod(targetPath, mode); err != nil {
                        return err
                }
        } else {
                klog.V(2).Infof("skip chmod on targetPath(%s) since mode is already 0%o)", targetPath, info.Mode())
        }
        return nil
}

func createStorageAccountSecret(account, key string) map[string]string {
        secret := make(map[string]string)
        secret[defaultSecretAccountName] = account
        secret[defaultSecretAccountKey] = key
        return secret
}

// setKeyValueInMap set key/value pair in map
// key in the map is case insensitive, if key already exists, overwrite existing value
func setKeyValueInMap(m map[string]string, key, value string) {
        if m == nil {
                return
        }
        for k := range m {
                if strings.EqualFold(k, key) {
                        m[k] = value
                        return
                }
        }
        m[key] = value
}

// getValueInMap get value from map by key
// key in the map is case insensitive
func getValueInMap(m map[string]string, key string) string {
        if m == nil {
                return ""
        }
        for k, v := range m {
                if strings.EqualFold(k, key) {
                        return v
                }
        }
        return ""
}

// replaceWithMap replace key with value for str
func replaceWithMap(str string, m map[string]string) string {
        for k, v := range m {
                if k != "" {
                        str = strings.ReplaceAll(str, k, v)
                }
        }
        return str
}

kubernetes-sigs / blob-csi-driver / 13390042311

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