13386305256

Committed 18 Feb 2025 08:27AM UTC coverage: 77.861%. Remained the same

Build # 13386305256

Build Type

Pull #1839

github

Committed by

andyzhangx

Commit Message

feat: optimize azcopy perf in volume cloning scenario

Pull Request Pull Request #1839: feat: optimize azcopy perf in volume cloning scenario

Run Details

2286 of 2936 relevant lines covered (77.86%)

7.44 hits per line

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

26.32

/pkg/blobplugin/main.go

/*
Copyright 2017 The Kubernetes Authors.

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

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

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

package main

import (
        "context"
        "flag"
        "fmt"
        "net"
        "net/http"
        "os"
        "runtime"
        "strings"

        "sigs.k8s.io/blob-csi-driver/pkg/blob"
        "sigs.k8s.io/blob-csi-driver/pkg/util"

        "k8s.io/component-base/metrics/legacyregistry"
        "k8s.io/klog/v2"
)

var driverOptions blob.DriverOptions
var (
        metricsAddress             = flag.String("metrics-address", "", "export the metrics")
        version                    = flag.Bool("version", false, "Print the version and exit.")
        endpoint                   = flag.String("endpoint", "unix://tmp/csi.sock", "CSI endpoint")
        kubeconfig                 = flag.String("kubeconfig", "", "Absolute path to the kubeconfig file. Required only when running out of cluster.")
        cloudConfigSecretName      = flag.String("cloud-config-secret-name", "azure-cloud-provider", "secret name of cloud config")
        cloudConfigSecretNamespace = flag.String("cloud-config-secret-namespace", "kube-system", "secret namespace of cloud config")
        customUserAgent            = flag.String("custom-user-agent", "", "custom userAgent")
        userAgentSuffix            = flag.String("user-agent-suffix", "", "userAgent suffix")
        allowEmptyCloudConfig      = flag.Bool("allow-empty-cloud-config", true, "allow running driver without cloud config")
        kubeAPIQPS                 = flag.Float64("kube-api-qps", 25.0, "QPS to use while communicating with the kubernetes apiserver.")
        kubeAPIBurst               = flag.Int("kube-api-burst", 50, "Burst to use while communicating with the kubernetes apiserver.")
        goMaxProcs                 = flag.Int("max-procs", 2, "maximum number of CPUs that can be executing simultaneously in golang runtime")
)

func init() {
        klog.InitFlags(nil)
        _ = flag.Set("logtostderr", "true")
        driverOptions.AddFlags()
}

// exit is a separate function to handle program termination
var exit = func(code int) {
        os.Exit(code)
}

func main() {
        flag.Parse()
        if *version {
                info, err := blob.GetVersionYAML(driverOptions.DriverName)
                if err != nil {
                        klog.Fatalln(err)
                }
                fmt.Println(info) // nolint
        } else {
                exportMetrics()
                handle()
        }
        exit(0)
}

func handle() {
        runtime.GOMAXPROCS(*goMaxProcs)
        klog.Infof("Sys info: NumCPU: %v MAXPROC: %v", runtime.NumCPU(), runtime.GOMAXPROCS(0))

        userAgent := blob.GetUserAgent(driverOptions.DriverName, *customUserAgent, *userAgentSuffix)
        klog.V(2).Infof("driver userAgent: %s", userAgent)

        kubeClient, err := util.GetKubeClient(*kubeconfig, *kubeAPIQPS, *kubeAPIBurst, userAgent)
        if err != nil {
                klog.Warningf("failed to get kubeClient, error: %v", err)
        }

        cloud, err := blob.GetCloudProvider(context.Background(), kubeClient, driverOptions.NodeID, *cloudConfigSecretName, *cloudConfigSecretNamespace, userAgent, *allowEmptyCloudConfig)
        if err != nil {
                klog.Fatalf("failed to get Azure Cloud Provider, error: %v", err)
        }
        klog.V(2).Infof("cloud: %s, location: %s, rg: %s, VnetName: %s, VnetResourceGroup: %s, SubnetName: %s", cloud.Cloud, cloud.Location, cloud.ResourceGroup, cloud.VnetName, cloud.VnetResourceGroup, cloud.SubnetName)

        driver := blob.NewDriver(&driverOptions, kubeClient, cloud)
        if driver == nil {
                klog.Fatalln("Failed to initialize Azure Blob Storage CSI driver")
        }
        if err := driver.Run(context.Background(), *endpoint); err != nil {
                klog.Fatalf("Failed to run Azure Blob Storage CSI driver: %v", err)
        }
}

func exportMetrics() {
        if *metricsAddress == "" {
                return
        }
        l, err := net.Listen("tcp", *metricsAddress)
        if err != nil {
                klog.Warningf("failed to get listener for metrics endpoint: %v", err)
                return
        }
        serve(context.Background(), l, serveMetrics)
}

func serve(_ context.Context, l net.Listener, serveFunc func(net.Listener) error) {
        path := l.Addr().String()
        klog.V(2).Infof("set up prometheus server on %v", path)
        go func() {
                defer l.Close()
                if err := serveFunc(l); err != nil {
                        klog.Fatalf("serve failure(%v), address(%v)", err, path)
                }
        }()
}

func serveMetrics(l net.Listener) error {
        m := http.NewServeMux()
        m.Handle("/metrics", legacyregistry.Handler()) //nolint, because azure cloud provider uses legacyregistry currently
        return trapClosedConnErr(http.Serve(l, m))
}

func trapClosedConnErr(err error) error {
        if err == nil {
                return nil
        }
        if strings.Contains(err.Error(), "use of closed network connection") {
                return nil
        }
        return err
}

1	/*
2	Copyright 2017 The Kubernetes Authors.
3
4	Licensed under the Apache License, Version 2.0 (the "License");
5	you may not use this file except in compliance with the License.
6	You may obtain a copy of the License at
7
8	http://www.apache.org/licenses/LICENSE-2.0
9
10	Unless required by applicable law or agreed to in writing, software
11	distributed under the License is distributed on an "AS IS" BASIS,
12	WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13	See the License for the specific language governing permissions and
14	limitations under the License.
15	*/
16
17	package main
18
19	import (
20	"context"
21	"flag"
22	"fmt"
23	"net"
24	"net/http"
25	"os"
26	"runtime"
27	"strings"
28
29	"sigs.k8s.io/blob-csi-driver/pkg/blob"
30	"sigs.k8s.io/blob-csi-driver/pkg/util"
31
32	"k8s.io/component-base/metrics/legacyregistry"
33	"k8s.io/klog/v2"
34	)
35
36	var driverOptions blob.DriverOptions
37	var (
38	metricsAddress = flag.String("metrics-address", "", "export the metrics")
39	version = flag.Bool("version", false, "Print the version and exit.")
40	endpoint = flag.String("endpoint", "unix://tmp/csi.sock", "CSI endpoint")
41	kubeconfig = flag.String("kubeconfig", "", "Absolute path to the kubeconfig file. Required only when running out of cluster.")
42	cloudConfigSecretName = flag.String("cloud-config-secret-name", "azure-cloud-provider", "secret name of cloud config")
43	cloudConfigSecretNamespace = flag.String("cloud-config-secret-namespace", "kube-system", "secret namespace of cloud config")
44	customUserAgent = flag.String("custom-user-agent", "", "custom userAgent")
45	userAgentSuffix = flag.String("user-agent-suffix", "", "userAgent suffix")
46	allowEmptyCloudConfig = flag.Bool("allow-empty-cloud-config", true, "allow running driver without cloud config")
47	kubeAPIQPS = flag.Float64("kube-api-qps", 25.0, "QPS to use while communicating with the kubernetes apiserver.")
48	kubeAPIBurst = flag.Int("kube-api-burst", 50, "Burst to use while communicating with the kubernetes apiserver.")
49	goMaxProcs = flag.Int("max-procs", 2, "maximum number of CPUs that can be executing simultaneously in golang runtime")
50	)
51
52	func init() {	1✔
53	klog.InitFlags(nil)	1✔
54	_ = flag.Set("logtostderr", "true")	1✔
55	driverOptions.AddFlags()	1✔
56	}	1✔
57
58	// exit is a separate function to handle program termination
59	var exit = func(code int) {	×
60	os.Exit(code)	×
61	}	×
62
63	func main() {	1✔
64	flag.Parse()	1✔
65	if *version {	2✔
66	info, err := blob.GetVersionYAML(driverOptions.DriverName)	1✔
67	if err != nil {	1✔
68	klog.Fatalln(err)	×
69	}	×
70	fmt.Println(info) // nolint	1✔
71	} else {	×
72	exportMetrics()	×
73	handle()	×
74	}	×
75	exit(0)	1✔
76	}
77
78	func handle() {	×
79	runtime.GOMAXPROCS(*goMaxProcs)	×
80	klog.Infof("Sys info: NumCPU: %v MAXPROC: %v", runtime.NumCPU(), runtime.GOMAXPROCS(0))	×
81		×
82	userAgent := blob.GetUserAgent(driverOptions.DriverName, customUserAgent, userAgentSuffix)	×
83	klog.V(2).Infof("driver userAgent: %s", userAgent)	×
84		×
85	kubeClient, err := util.GetKubeClient(kubeconfig, kubeAPIQPS, *kubeAPIBurst, userAgent)	×
86	if err != nil {	×
87	klog.Warningf("failed to get kubeClient, error: %v", err)	×
88	}	×
89
90	cloud, err := blob.GetCloudProvider(context.Background(), kubeClient, driverOptions.NodeID, cloudConfigSecretName, cloudConfigSecretNamespace, userAgent, *allowEmptyCloudConfig)	×
91	if err != nil {	×
92	klog.Fatalf("failed to get Azure Cloud Provider, error: %v", err)	×
93	}	×
94	klog.V(2).Infof("cloud: %s, location: %s, rg: %s, VnetName: %s, VnetResourceGroup: %s, SubnetName: %s", cloud.Cloud, cloud.Location, cloud.ResourceGroup, cloud.VnetName, cloud.VnetResourceGroup, cloud.SubnetName)	×
95		×
96	driver := blob.NewDriver(&driverOptions, kubeClient, cloud)	×
97	if driver == nil {	×
98	klog.Fatalln("Failed to initialize Azure Blob Storage CSI driver")	×
99	}	×
100	if err := driver.Run(context.Background(), *endpoint); err != nil {	×
101	klog.Fatalf("Failed to run Azure Blob Storage CSI driver: %v", err)	×
102	}	×
103	}
104
105	func exportMetrics() {	×
106	if *metricsAddress == "" {	×
107	return	×
108	}	×
109	l, err := net.Listen("tcp", *metricsAddress)	×
110	if err != nil {	×
111	klog.Warningf("failed to get listener for metrics endpoint: %v", err)	×
112	return	×
113	}	×
114	serve(context.Background(), l, serveMetrics)	×
115	}
116
117	func serve(_ context.Context, l net.Listener, serveFunc func(net.Listener) error) {	×
118	path := l.Addr().String()	×
119	klog.V(2).Infof("set up prometheus server on %v", path)	×
120	go func() {	×
121	defer l.Close()	×
122	if err := serveFunc(l); err != nil {	×
123	klog.Fatalf("serve failure(%v), address(%v)", err, path)	×
124	}	×
125	}()
126	}
127
128	func serveMetrics(l net.Listener) error {	×
129	m := http.NewServeMux()	×
130	m.Handle("/metrics", legacyregistry.Handler()) //nolint, because azure cloud provider uses legacyregistry currently	×
131	return trapClosedConnErr(http.Serve(l, m))	×
132	}	×
133
134	func trapClosedConnErr(err error) error {	3✔
135	if err == nil {	4✔
136	return nil	1✔
137	}	1✔
138	if strings.Contains(err.Error(), "use of closed network connection") {	3✔
139	return nil	1✔
140	}	1✔
141	return err	1✔
142	}

kubernetes-sigs / blob-csi-driver / 13386305256

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