9289598898

Committed 29 May 2024 04:17PM UTC coverage: 74.251%. Remained the same

Build # 9289598898

Build Type

Pull #1406

github

Committed by

web-flow

Commit Message

chore(deps): bump sigs.k8s.io/cloud-provider-azure/pkg/azclient

Bumps [sigs.k8s.io/cloud-provider-azure/pkg/azclient](https://github.com/kubernetes-sigs/cloud-provider-azure) from 0.0.21 to 0.0.23.
- [Release notes](https://github.com/kubernetes-sigs/cloud-provider-azure/releases)
- [Changelog](https://github.com/kubernetes-sigs/cloud-provider-azure/blob/master/docs/release-versioning.md)
- [Commits](https://github.com/kubernetes-sigs/cloud-provider-azure/compare/pkg/azclient/v0.0.21...pkg/azclient/v0.0.23)

---
updated-dependencies:
- dependency-name: sigs.k8s.io/cloud-provider-azure/pkg/azclient
  dependency-type: direct:production
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>

Pull Request Pull Request #1406: chore(deps): bump sigs.k8s.io/cloud-provider-azure/pkg/azclient from 0.0.21 to 0.0.23

Run Details

2206 of 2971 relevant lines covered (74.25%)

7.09 hits per line

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

85.48

/pkg/util/util.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 util

import (
        "fmt"
        "os"
        "os/exec"
        "regexp"
        "strconv"
        "strings"
        "sync"
        "time"

        "github.com/go-ini/ini"
        "github.com/pkg/errors"
        v1 "k8s.io/api/core/v1"
        "k8s.io/client-go/kubernetes"
        "k8s.io/client-go/rest"
        "k8s.io/client-go/tools/clientcmd"
        "k8s.io/klog/v2"
        "k8s.io/kubernetes/pkg/volume"
)

const (
        GiB                  = 1024 * 1024 * 1024
        TiB                  = 1024 * GiB
        tagsDelimiter        = ","
        tagKeyValueDelimiter = "="
)

type AzcopyJobState string

const (
        AzcopyJobError     AzcopyJobState = "Error"
        AzcopyJobNotFound  AzcopyJobState = "NotFound"
        AzcopyJobRunning   AzcopyJobState = "Running"
        AzcopyJobCompleted AzcopyJobState = "Completed"
)

// RoundUpBytes rounds up the volume size in bytes up to multiplications of GiB
// in the unit of Bytes
func RoundUpBytes(volumeSizeBytes int64) int64 {
        return roundUpSize(volumeSizeBytes, GiB) * GiB
}

// RoundUpGiB rounds up the volume size in bytes up to multiplications of GiB
// in the unit of GiB
func RoundUpGiB(volumeSizeBytes int64) int64 {
        return roundUpSize(volumeSizeBytes, GiB)
}

// BytesToGiB conversts Bytes to GiB
func BytesToGiB(volumeSizeBytes int64) int64 {
        return volumeSizeBytes / GiB
}

// GiBToBytes converts GiB to Bytes
func GiBToBytes(volumeSizeGiB int64) int64 {
        return volumeSizeGiB * GiB
}

// roundUpSize calculates how many allocation units are needed to accommodate
// a volume of given size. E.g. when user wants 1500MiB volume, while AWS EBS
// allocates volumes in gibibyte-sized chunks,
// RoundUpSize(1500 * 1024*1024, 1024*1024*1024) returns '2'
// (2 GiB is the smallest allocatable volume that can hold 1500MiB)
func roundUpSize(volumeSizeBytes int64, allocationUnitBytes int64) int64 {
        roundedUp := volumeSizeBytes / allocationUnitBytes
        if volumeSizeBytes%allocationUnitBytes > 0 {
                roundedUp++
        }
        return roundedUp
}

// GetMountOptions return options with string list separated by space
func GetMountOptions(options []string) string {
        if len(options) == 0 {
                return ""
        }
        str := options[0]
        for i := 1; i < len(options); i++ {
                str = str + " " + options[i]
        }
        return str
}

func MakeDir(pathname string, perm os.FileMode) error {
        err := os.MkdirAll(pathname, perm)
        if err != nil {
                if !os.IsExist(err) {
                        return err
                }
        }
        return nil
}

// LockMap used to lock on entries
type LockMap struct {
        sync.Mutex
        mutexMap map[string]*sync.Mutex
}

// NewLockMap returns a new lock map
func NewLockMap() *LockMap {
        return &LockMap{
                mutexMap: make(map[string]*sync.Mutex),
        }
}

// LockEntry acquires a lock associated with the specific entry
func (lm *LockMap) LockEntry(entry string) {
        lm.Lock()
        // check if entry does not exists, then add entry
        if _, exists := lm.mutexMap[entry]; !exists {
                lm.addEntry(entry)
        }

        lm.Unlock()
        lm.lockEntry(entry)
}

// UnlockEntry release the lock associated with the specific entry
func (lm *LockMap) UnlockEntry(entry string) {
        lm.Lock()
        defer lm.Unlock()

        if _, exists := lm.mutexMap[entry]; !exists {
                return
        }
        lm.unlockEntry(entry)
}

func (lm *LockMap) addEntry(entry string) {
        lm.mutexMap[entry] = &sync.Mutex{}
}

func (lm *LockMap) lockEntry(entry string) {
        lm.mutexMap[entry].Lock()
}

func (lm *LockMap) unlockEntry(entry string) {
        lm.mutexMap[entry].Unlock()
}

func ConvertTagsToMap(tags string) (map[string]string, error) {
        m := make(map[string]string)
        if tags == "" {
                return m, nil
        }
        s := strings.Split(tags, tagsDelimiter)
        for _, tag := range s {
                kv := strings.Split(tag, tagKeyValueDelimiter)
                if len(kv) != 2 {
                        return nil, fmt.Errorf("Tags '%s' are invalid, the format should like: 'key1=value1,key2=value2'", tags)
                }
                key := strings.TrimSpace(kv[0])
                if key == "" {
                        return nil, fmt.Errorf("Tags '%s' are invalid, the format should like: 'key1=value1,key2=value2'", tags)
                }
                value := strings.TrimSpace(kv[1])
                m[key] = value
        }
        return m, nil
}

type OsInfo struct {
        Distro  string
        Version string
}

const (
        keyID        = "ID"
        keyVersionID = "VERSION_ID"
)

func GetOSInfo(f interface{}) (*OsInfo, error) {
        cfg, err := ini.Load(f)
        if err != nil {
                return nil, errors.Wrapf(err, "failed to read %q", f)
        }

        oi := &OsInfo{}
        oi.Distro = cfg.Section("").Key(keyID).String()
        oi.Version = cfg.Section("").Key(keyVersionID).String()

        klog.V(2).Infof("get OS info: %v", oi)
        return oi, nil
}

func TrimDuplicatedSpace(s string) string {
        reg := regexp.MustCompile(`\s+`)
        s = reg.ReplaceAllString(s, " ")
        return s
}

type EXEC interface {
        RunCommand(string, []string) (string, error)
}

type ExecCommand struct {
}

func (ec *ExecCommand) RunCommand(cmdStr string, authEnv []string) (string, error) {
        cmd := exec.Command("sh", "-c", cmdStr)
        if len(authEnv) > 0 {
                cmd.Env = append(os.Environ(), authEnv...)
        }
        out, err := cmd.CombinedOutput()
        return string(out), err
}

type Azcopy struct {
        ExecCmd EXEC
}

// GetAzcopyJob get the azcopy job status if job existed
func (ac *Azcopy) GetAzcopyJob(dstBlobContainer string, authAzcopyEnv []string) (AzcopyJobState, string, error) {
        cmdStr := fmt.Sprintf("azcopy jobs list | grep %s -B 3", dstBlobContainer)
        // cmd output example:
        // JobId: ed1c3833-eaff-fe42-71d7-513fb065a9d9
        // Start Time: Monday, 07-Aug-23 03:29:54 UTC
        // Status: Completed (or Cancelled, InProgress)
        // Command: copy https://{accountName}.file.core.windows.net/{srcBlobContainer}{SAStoken} https://{accountName}.file.core.windows.net/{dstBlobContainer}{SAStoken} --recursive --check-length=false
        // --
        // JobId: b598cce3-9aa9-9640-7793-c2bf3c385a9a
        // Start Time: Wednesday, 09-Aug-23 09:09:03 UTC
        // Status: Cancelled
        // Command: copy https://{accountName}.file.core.windows.net/{srcBlobContainer}{SAStoken} https://{accountName}.file.core.windows.net/{dstBlobContainer}{SAStoken} --recursive --check-length=false
        if ac.ExecCmd == nil {
                ac.ExecCmd = &ExecCommand{}
        }
        out, err := ac.ExecCmd.RunCommand(cmdStr, authAzcopyEnv)
        // if grep command returns nothing, the exec will return exit status 1 error, so filter this error
        if err != nil && err.Error() != "exit status 1" {
                klog.Warningf("failed to get azcopy job with error: %v, jobState: %v", err, AzcopyJobError)
                return AzcopyJobError, "", fmt.Errorf("couldn't list jobs in azcopy %v", err)
        }
        jobid, jobState, err := parseAzcopyJobList(out)
        if err != nil || jobState == AzcopyJobError {
                klog.Warningf("failed to get azcopy job with error: %v, jobState: %v", err, jobState)
                return AzcopyJobError, "", fmt.Errorf("couldn't parse azcopy job list in azcopy %v", err)
        }
        if jobState == AzcopyJobCompleted {
                return jobState, "100.0", err
        }
        if jobid == "" {
                return jobState, "", err
        }
        cmdPercentStr := fmt.Sprintf("azcopy jobs show %s | grep Percent", jobid)
        // cmd out example:
        // Percent Complete (approx): 100.0
        summary, err := ac.ExecCmd.RunCommand(cmdPercentStr, authAzcopyEnv)
        if err != nil {
                klog.Warningf("failed to get azcopy job with error: %v, jobState: %v", err, AzcopyJobError)
                return AzcopyJobError, "", fmt.Errorf("couldn't show jobs summary in azcopy %v", err)
        }
        jobState, percent, err := parseAzcopyJobShow(summary)
        if err != nil || jobState == AzcopyJobError {
                klog.Warningf("failed to get azcopy job with error: %v, jobState: %v", err, jobState)
                return AzcopyJobError, "", fmt.Errorf("couldn't parse azcopy job show in azcopy %v", err)
        }
        return jobState, percent, nil
}

// TestListJobs test azcopy jobs list command with authAzcopyEnv
func (ac *Azcopy) TestListJobs(accountName, storageEndpointSuffix string, authAzcopyEnv []string) (string, error) {
        cmdStr := fmt.Sprintf("azcopy list %s", fmt.Sprintf("https://%s.blob.%s", accountName, storageEndpointSuffix))
        if ac.ExecCmd == nil {
                ac.ExecCmd = &ExecCommand{}
        }
        return ac.ExecCmd.RunCommand(cmdStr, authAzcopyEnv)
}

// parseAzcopyJobList parse command azcopy jobs list, get jobid and state from joblist
func parseAzcopyJobList(joblist string) (string, AzcopyJobState, error) {
        jobid := ""
        jobSegments := strings.Split(joblist, "JobId: ")
        if len(jobSegments) < 2 {
                return jobid, AzcopyJobNotFound, nil
        }
        jobSegments = jobSegments[1:]
        for _, job := range jobSegments {
                segments := strings.Split(job, "\n")
                if len(segments) < 4 {
                        return jobid, AzcopyJobError, fmt.Errorf("error parsing jobs list: %s", job)
                }
                statusSegments := strings.Split(segments[2], ": ")
                if len(statusSegments) < 2 {
                        return jobid, AzcopyJobError, fmt.Errorf("error parsing jobs list status: %s", segments[2])
                }
                status := statusSegments[1]
                switch status {
                case "InProgress":
                        jobid = segments[0]
                case "Completed":
                        return jobid, AzcopyJobCompleted, nil
                }
        }
        if jobid == "" {
                return jobid, AzcopyJobNotFound, nil
        }
        return jobid, AzcopyJobRunning, nil
}

// parseAzcopyJobShow parse command azcopy jobs show jobid, get job state and copy percent
func parseAzcopyJobShow(jobshow string) (AzcopyJobState, string, error) {
        segments := strings.Split(jobshow, ": ")
        if len(segments) < 2 {
                return AzcopyJobError, "", fmt.Errorf("error parsing jobs summary: %s in Percent Complete (approx)", jobshow)
        }
        return AzcopyJobRunning, strings.ReplaceAll(segments[1], "\n", ""), nil
}

func GetKubeClient(kubeconfig string, kubeAPIQPS float64, kubeAPIBurst int, userAgent string) (kubernetes.Interface, error) {
        var err error
        var kubeCfg *rest.Config
        if kubeCfg, err = clientcmd.BuildConfigFromFlags("", kubeconfig); err != nil {
                return nil, err
        }
        if kubeCfg == nil {
                if kubeCfg, err = rest.InClusterConfig(); err != nil {
                        return nil, err
                }
        }
        //kubeCfg should not be nil
        // set QPS and QPS Burst as higher values
        klog.V(2).Infof("set QPS(%f) and QPS Burst(%d) for driver kubeClient", float32(kubeAPIQPS), kubeAPIBurst)
        kubeCfg.QPS = float32(kubeAPIQPS)
        kubeCfg.Burst = kubeAPIBurst
        kubeCfg.UserAgent = userAgent
        return kubernetes.NewForConfig(kubeCfg)
}

type VolumeMounter struct {
        path       string
        attributes volume.Attributes
}

func (l *VolumeMounter) GetPath() string {
        return l.path
}

func (l *VolumeMounter) GetAttributes() volume.Attributes {
        return l.attributes
}

func (l *VolumeMounter) CanMount() error {
        return nil
}

func (l *VolumeMounter) SetUp(_ volume.MounterArgs) error {
        return nil
}

func (l *VolumeMounter) SetUpAt(_ string, _ volume.MounterArgs) error {
        return nil
}

func (l *VolumeMounter) GetMetrics() (*volume.Metrics, error) {
        return nil, nil
}

// SetVolumeOwnership would set gid for path recursively
func SetVolumeOwnership(path, gid, policy string) error {
        id, err := strconv.Atoi(gid)
        if err != nil {
                return fmt.Errorf("convert %s to int failed with %v", gid, err)
        }
        gidInt64 := int64(id)
        fsGroupChangePolicy := v1.FSGroupChangeOnRootMismatch
        if policy != "" {
                fsGroupChangePolicy = v1.PodFSGroupChangePolicy(policy)
        }
        return volume.SetVolumeOwnership(&VolumeMounter{path: path}, path, &gidInt64, &fsGroupChangePolicy, nil)
}

// ExecFunc returns a exec function's output and error
type ExecFunc func() (err error)

// TimeoutFunc returns output and error if an ExecFunc timeout
type TimeoutFunc func() (err error)

// WaitUntilTimeout waits for the exec function to complete or return timeout error
func WaitUntilTimeout(timeout time.Duration, execFunc ExecFunc, timeoutFunc TimeoutFunc) error {
        // Create a channel to receive the result of the azcopy exec function
        done := make(chan bool)
        var err error

        // Start the azcopy exec function in a goroutine
        go func() {
                err = execFunc()
                done <- true
        }()

        // Wait for the function to complete or time out
        select {
        case <-done:
                return err
        case <-time.After(timeout):
                return timeoutFunc()
        }
}

1	/*
2	Copyright 2019 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 util
18
19	import (
20	"fmt"
21	"os"
22	"os/exec"
23	"regexp"
24	"strconv"
25	"strings"
26	"sync"
27	"time"
28
29	"github.com/go-ini/ini"
30	"github.com/pkg/errors"
31	v1 "k8s.io/api/core/v1"
32	"k8s.io/client-go/kubernetes"
33	"k8s.io/client-go/rest"
34	"k8s.io/client-go/tools/clientcmd"
35	"k8s.io/klog/v2"
36	"k8s.io/kubernetes/pkg/volume"
37	)
38
39	const (
40	GiB = 1024 * 1024 * 1024
41	TiB = 1024 * GiB
42	tagsDelimiter = ","
43	tagKeyValueDelimiter = "="
44	)
45
46	type AzcopyJobState string
47
48	const (
49	AzcopyJobError AzcopyJobState = "Error"
50	AzcopyJobNotFound AzcopyJobState = "NotFound"
51	AzcopyJobRunning AzcopyJobState = "Running"
52	AzcopyJobCompleted AzcopyJobState = "Completed"
53	)
54
55	// RoundUpBytes rounds up the volume size in bytes up to multiplications of GiB
56	// in the unit of Bytes
57	func RoundUpBytes(volumeSizeBytes int64) int64 {	1✔
58	return roundUpSize(volumeSizeBytes, GiB) * GiB	1✔
59	}	1✔
60
61	// RoundUpGiB rounds up the volume size in bytes up to multiplications of GiB
62	// in the unit of GiB
63	func RoundUpGiB(volumeSizeBytes int64) int64 {	1✔
64	return roundUpSize(volumeSizeBytes, GiB)	1✔
65	}	1✔
66
67	// BytesToGiB conversts Bytes to GiB
68	func BytesToGiB(volumeSizeBytes int64) int64 {	1✔
69	return volumeSizeBytes / GiB	1✔
70	}	1✔
71
72	// GiBToBytes converts GiB to Bytes
73	func GiBToBytes(volumeSizeGiB int64) int64 {	1✔
74	return volumeSizeGiB * GiB	1✔
75	}	1✔
76
77	// roundUpSize calculates how many allocation units are needed to accommodate
78	// a volume of given size. E.g. when user wants 1500MiB volume, while AWS EBS
79	// allocates volumes in gibibyte-sized chunks,
80	// RoundUpSize(1500 * 10241024, 10241024*1024) returns '2'
81	// (2 GiB is the smallest allocatable volume that can hold 1500MiB)
82	func roundUpSize(volumeSizeBytes int64, allocationUnitBytes int64) int64 {	2✔
83	roundedUp := volumeSizeBytes / allocationUnitBytes	2✔
84	if volumeSizeBytes%allocationUnitBytes > 0 {	4✔
85	roundedUp++	2✔
86	}	2✔
87	return roundedUp	2✔
88	}
89
90	// GetMountOptions return options with string list separated by space
91	func GetMountOptions(options []string) string {	4✔
92	if len(options) == 0 {	5✔
93	return ""	1✔
94	}	1✔
95	str := options[0]	3✔
96	for i := 1; i < len(options); i++ {	5✔
97	str = str + " " + options[i]	2✔
98	}	2✔
99	return str	3✔
100	}
101
102	func MakeDir(pathname string, perm os.FileMode) error {	1✔
103	err := os.MkdirAll(pathname, perm)	1✔
104	if err != nil {	1✔
105	if !os.IsExist(err) {	×
106	return err	×
107	}	×
108	}
109	return nil	1✔
110	}
111
112	// LockMap used to lock on entries
113	type LockMap struct {
114	sync.Mutex
115	mutexMap map[string]*sync.Mutex
116	}
117
118	// NewLockMap returns a new lock map
119	func NewLockMap() *LockMap {	4✔
120	return &LockMap{	4✔
121	mutexMap: make(map[string]*sync.Mutex),	4✔
122	}	4✔
123	}	4✔
124
125	// LockEntry acquires a lock associated with the specific entry
126	func (lm *LockMap) LockEntry(entry string) {	5✔
127	lm.Lock()	5✔
128	// check if entry does not exists, then add entry	5✔
129	if _, exists := lm.mutexMap[entry]; !exists {	9✔
130	lm.addEntry(entry)	4✔
131	}	4✔
132
133	lm.Unlock()	5✔
134	lm.lockEntry(entry)	5✔
135	}
136
137	// UnlockEntry release the lock associated with the specific entry
138	func (lm *LockMap) UnlockEntry(entry string) {	5✔
139	lm.Lock()	5✔
140	defer lm.Unlock()	5✔
141		5✔
142	if _, exists := lm.mutexMap[entry]; !exists {	6✔
143	return	1✔
144	}	1✔
145	lm.unlockEntry(entry)	4✔
146	}
147
148	func (lm *LockMap) addEntry(entry string) {	4✔
149	lm.mutexMap[entry] = &sync.Mutex{}	4✔
150	}	4✔
151
152	func (lm *LockMap) lockEntry(entry string) {	5✔
153	lm.mutexMap[entry].Lock()	5✔
154	}	5✔
155
156	func (lm *LockMap) unlockEntry(entry string) {	4✔
157	lm.mutexMap[entry].Unlock()	4✔
158	}	4✔
159
160	func ConvertTagsToMap(tags string) (map[string]string, error) {	8✔
161	m := make(map[string]string)	8✔
162	if tags == "" {	9✔
163	return m, nil	1✔
164	}	1✔
165	s := strings.Split(tags, tagsDelimiter)	7✔
166	for _, tag := range s {	17✔
167	kv := strings.Split(tag, tagKeyValueDelimiter)	10✔
168	if len(kv) != 2 {	12✔
169	return nil, fmt.Errorf("Tags '%s' are invalid, the format should like: 'key1=value1,key2=value2'", tags)	2✔
170	}	2✔
171	key := strings.TrimSpace(kv[0])	8✔
172	if key == "" {	10✔
173	return nil, fmt.Errorf("Tags '%s' are invalid, the format should like: 'key1=value1,key2=value2'", tags)	2✔
174	}	2✔
175	value := strings.TrimSpace(kv[1])	6✔
176	m[key] = value	6✔
177	}
178	return m, nil	3✔
179	}
180
181	type OsInfo struct {
182	Distro string
183	Version string
184	}
185
186	const (
187	keyID = "ID"
188	keyVersionID = "VERSION_ID"
189	)
190
191	func GetOSInfo(f interface{}) (*OsInfo, error) {	2✔
192	cfg, err := ini.Load(f)	2✔
193	if err != nil {	3✔
194	return nil, errors.Wrapf(err, "failed to read %q", f)	1✔
195	}	1✔
196
197	oi := &OsInfo{}	1✔
198	oi.Distro = cfg.Section("").Key(keyID).String()	1✔
199	oi.Version = cfg.Section("").Key(keyVersionID).String()	1✔
200		1✔
201	klog.V(2).Infof("get OS info: %v", oi)	1✔
202	return oi, nil	1✔
203	}
204
205	func TrimDuplicatedSpace(s string) string {	1✔
206	reg := regexp.MustCompile(`\s+`)	1✔
207	s = reg.ReplaceAllString(s, " ")	1✔
208	return s	1✔
209	}	1✔
210
211	type EXEC interface {
212	RunCommand(string, []string) (string, error)
213	}
214
215	type ExecCommand struct {
216	}
217
218	func (ec *ExecCommand) RunCommand(cmdStr string, authEnv []string) (string, error) {	×
219	cmd := exec.Command("sh", "-c", cmdStr)	×
220	if len(authEnv) > 0 {	×
221	cmd.Env = append(os.Environ(), authEnv...)	×
222	}	×
223	out, err := cmd.CombinedOutput()	×
224	return string(out), err	×
225	}
226
227	type Azcopy struct {
228	ExecCmd EXEC
229	}
230
231	// GetAzcopyJob get the azcopy job status if job existed
232	func (ac *Azcopy) GetAzcopyJob(dstBlobContainer string, authAzcopyEnv []string) (AzcopyJobState, string, error) {	7✔
233	cmdStr := fmt.Sprintf("azcopy jobs list \| grep %s -B 3", dstBlobContainer)	7✔
234	// cmd output example:	7✔
235	// JobId: ed1c3833-eaff-fe42-71d7-513fb065a9d9	7✔
236	// Start Time: Monday, 07-Aug-23 03:29:54 UTC	7✔
237	// Status: Completed (or Cancelled, InProgress)	7✔
238	// Command: copy https://{accountName}.file.core.windows.net/{srcBlobContainer}{SAStoken} https://{accountName}.file.core.windows.net/{dstBlobContainer}{SAStoken} --recursive --check-length=false	7✔
239	// --	7✔
240	// JobId: b598cce3-9aa9-9640-7793-c2bf3c385a9a	7✔
241	// Start Time: Wednesday, 09-Aug-23 09:09:03 UTC	7✔
242	// Status: Cancelled	7✔
243	// Command: copy https://{accountName}.file.core.windows.net/{srcBlobContainer}{SAStoken} https://{accountName}.file.core.windows.net/{dstBlobContainer}{SAStoken} --recursive --check-length=false	7✔
244	if ac.ExecCmd == nil {	7✔
245	ac.ExecCmd = &ExecCommand{}	×
246	}	×
247	out, err := ac.ExecCmd.RunCommand(cmdStr, authAzcopyEnv)	7✔
248	// if grep command returns nothing, the exec will return exit status 1 error, so filter this error	7✔
249	if err != nil && err.Error() != "exit status 1" {	8✔
250	klog.Warningf("failed to get azcopy job with error: %v, jobState: %v", err, AzcopyJobError)	1✔
251	return AzcopyJobError, "", fmt.Errorf("couldn't list jobs in azcopy %v", err)	1✔
252	}	1✔
253	jobid, jobState, err := parseAzcopyJobList(out)	6✔
254	if err != nil \|\| jobState == AzcopyJobError {	7✔
255	klog.Warningf("failed to get azcopy job with error: %v, jobState: %v", err, jobState)	1✔
256	return AzcopyJobError, "", fmt.Errorf("couldn't parse azcopy job list in azcopy %v", err)	1✔
257	}	1✔
258	if jobState == AzcopyJobCompleted {	6✔
259	return jobState, "100.0", err	1✔
260	}	1✔
261	if jobid == "" {	5✔
262	return jobState, "", err	1✔
263	}	1✔
264	cmdPercentStr := fmt.Sprintf("azcopy jobs show %s \| grep Percent", jobid)	3✔
265	// cmd out example:	3✔
266	// Percent Complete (approx): 100.0	3✔
267	summary, err := ac.ExecCmd.RunCommand(cmdPercentStr, authAzcopyEnv)	3✔
268	if err != nil {	4✔
269	klog.Warningf("failed to get azcopy job with error: %v, jobState: %v", err, AzcopyJobError)	1✔
270	return AzcopyJobError, "", fmt.Errorf("couldn't show jobs summary in azcopy %v", err)	1✔
271	}	1✔
272	jobState, percent, err := parseAzcopyJobShow(summary)	2✔
273	if err != nil \|\| jobState == AzcopyJobError {	3✔
274	klog.Warningf("failed to get azcopy job with error: %v, jobState: %v", err, jobState)	1✔
275	return AzcopyJobError, "", fmt.Errorf("couldn't parse azcopy job show in azcopy %v", err)	1✔
276	}	1✔
277	return jobState, percent, nil	1✔
278	}
279
280	// TestListJobs test azcopy jobs list command with authAzcopyEnv
281	func (ac *Azcopy) TestListJobs(accountName, storageEndpointSuffix string, authAzcopyEnv []string) (string, error) {	×
282	cmdStr := fmt.Sprintf("azcopy list %s", fmt.Sprintf("https://%s.blob.%s", accountName, storageEndpointSuffix))	×
283	if ac.ExecCmd == nil {	×
284	ac.ExecCmd = &ExecCommand{}	×
285	}	×
286	return ac.ExecCmd.RunCommand(cmdStr, authAzcopyEnv)	×
287	}
288
289	// parseAzcopyJobList parse command azcopy jobs list, get jobid and state from joblist
290	func parseAzcopyJobList(joblist string) (string, AzcopyJobState, error) {	12✔
291	jobid := ""	12✔
292	jobSegments := strings.Split(joblist, "JobId: ")	12✔
293	if len(jobSegments) < 2 {	13✔
294	return jobid, AzcopyJobNotFound, nil	1✔
295	}	1✔
296	jobSegments = jobSegments[1:]	11✔
297	for _, job := range jobSegments {	22✔
298	segments := strings.Split(job, "\n")	11✔
299	if len(segments) < 4 {	13✔
300	return jobid, AzcopyJobError, fmt.Errorf("error parsing jobs list: %s", job)	2✔
301	}	2✔
302	statusSegments := strings.Split(segments[2], ": ")	9✔
303	if len(statusSegments) < 2 {	10✔
304	return jobid, AzcopyJobError, fmt.Errorf("error parsing jobs list status: %s", segments[2])	1✔
305	}	1✔
306	status := statusSegments[1]	8✔
307	switch status {	8✔
308	case "InProgress":	4✔
309	jobid = segments[0]	4✔
310	case "Completed":	2✔
311	return jobid, AzcopyJobCompleted, nil	2✔
312	}
313	}
314	if jobid == "" {	8✔
315	return jobid, AzcopyJobNotFound, nil	2✔
316	}	2✔
317	return jobid, AzcopyJobRunning, nil	4✔
318	}
319
320	// parseAzcopyJobShow parse command azcopy jobs show jobid, get job state and copy percent
321	func parseAzcopyJobShow(jobshow string) (AzcopyJobState, string, error) {	4✔
322	segments := strings.Split(jobshow, ": ")	4✔
323	if len(segments) < 2 {	6✔
324	return AzcopyJobError, "", fmt.Errorf("error parsing jobs summary: %s in Percent Complete (approx)", jobshow)	2✔
325	}	2✔
326	return AzcopyJobRunning, strings.ReplaceAll(segments[1], "\n", ""), nil	2✔
327	}
328
329	func GetKubeClient(kubeconfig string, kubeAPIQPS float64, kubeAPIBurst int, userAgent string) (kubernetes.Interface, error) {	3✔
330	var err error	3✔
331	var kubeCfg *rest.Config	3✔
332	if kubeCfg, err = clientcmd.BuildConfigFromFlags("", kubeconfig); err != nil {	5✔
333	return nil, err	2✔
334	}	2✔
335	if kubeCfg == nil {	1✔
336	if kubeCfg, err = rest.InClusterConfig(); err != nil {	×
337	return nil, err	×
338	}	×
339	}
340	//kubeCfg should not be nil
341	// set QPS and QPS Burst as higher values
342	klog.V(2).Infof("set QPS(%f) and QPS Burst(%d) for driver kubeClient", float32(kubeAPIQPS), kubeAPIBurst)	1✔
343	kubeCfg.QPS = float32(kubeAPIQPS)	1✔
344	kubeCfg.Burst = kubeAPIBurst	1✔
345	kubeCfg.UserAgent = userAgent	1✔
346	return kubernetes.NewForConfig(kubeCfg)	1✔
347	}
348
349	type VolumeMounter struct {
350	path string
351	attributes volume.Attributes
352	}
353
354	func (l *VolumeMounter) GetPath() string {	×
355	return l.path	×
356	}	×
357
358	func (l *VolumeMounter) GetAttributes() volume.Attributes {	6✔
359	return l.attributes	6✔
360	}	6✔
361
362	func (l *VolumeMounter) CanMount() error {	×
363	return nil	×
364	}	×
365
366	func (l *VolumeMounter) SetUp(_ volume.MounterArgs) error {	×
367	return nil	×
368	}	×
369
370	func (l *VolumeMounter) SetUpAt(_ string, _ volume.MounterArgs) error {	×
371	return nil	×
372	}	×
373
374	func (l VolumeMounter) GetMetrics() (volume.Metrics, error) {	×
375	return nil, nil	×
376	}	×
377
378	// SetVolumeOwnership would set gid for path recursively
379	func SetVolumeOwnership(path, gid, policy string) error {	6✔
380	id, err := strconv.Atoi(gid)	6✔
381	if err != nil {	8✔
382	return fmt.Errorf("convert %s to int failed with %v", gid, err)	2✔
383	}	2✔
384	gidInt64 := int64(id)	4✔
385	fsGroupChangePolicy := v1.FSGroupChangeOnRootMismatch	4✔
386	if policy != "" {	6✔
387	fsGroupChangePolicy = v1.PodFSGroupChangePolicy(policy)	2✔
388	}	2✔
389	return volume.SetVolumeOwnership(&VolumeMounter{path: path}, path, &gidInt64, &fsGroupChangePolicy, nil)	4✔
390	}
391
392	// ExecFunc returns a exec function's output and error
393	type ExecFunc func() (err error)
394
395	// TimeoutFunc returns output and error if an ExecFunc timeout
396	type TimeoutFunc func() (err error)
397
398	// WaitUntilTimeout waits for the exec function to complete or return timeout error
399	func WaitUntilTimeout(timeout time.Duration, execFunc ExecFunc, timeoutFunc TimeoutFunc) error {	3✔
400	// Create a channel to receive the result of the azcopy exec function	3✔
401	done := make(chan bool)	3✔
402	var err error	3✔
403		3✔
404	// Start the azcopy exec function in a goroutine	3✔
405	go func() {	6✔
406	err = execFunc()	3✔
407	done <- true	3✔
408	}()	3✔
409
410	// Wait for the function to complete or time out
411	select {	3✔
412	case <-done:	2✔
413	return err	2✔
414	case <-time.After(timeout):	1✔
415	return timeoutFunc()	1✔
416	}
417	}

kubernetes-sigs / blob-csi-driver / 9289598898

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