7412225870

Committed 04 Jan 2024 04:35PM UTC coverage: 77.871%. Remained the same

Build # 7412225870

Build Type

Pull #1209

github

Committed by

web-flow

Commit Message

chore(deps): bump golang.org/x/sync from 0.5.0 to 0.6.0

Bumps [golang.org/x/sync](https://github.com/golang/sync) from 0.5.0 to 0.6.0.
- [Commits](https://github.com/golang/sync/compare/v0.5.0...v0.6.0)

---
updated-dependencies:
- dependency-name: golang.org/x/sync
  dependency-type: direct:production
  update-type: version-update:semver-minor
...

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

Pull Request Pull Request #1209: chore(deps): bump golang.org/x/sync from 0.5.0 to 0.6.0

Run Details

2034 of 2612 relevant lines covered (77.87%)

7.29 hits per line

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

84.55

/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"

        "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)
}

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

kubernetes-sigs / blob-csi-driver / 7412225870

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