16064476868

Committed 04 Jul 2025 02:25AM UTC coverage: 78.36%. Remained the same

Build # 16064476868

Build Type

Pull #2058

github

Committed by

andyzhangx

Commit Message

chore: fix helm chart index on release-1.26

Pull Request Pull Request #2058: [release-1.26] chore: fix helm chart index on release-1.26

Run Details

2379 of 3036 relevant lines covered (78.36%)

7.65 hits per line

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

60.24

/pkg/blobfuse-proxy/server/server.go

/*
Copyright 2021 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 server

import (
        "context"
        "fmt"
        "net"
        "os"
        "os/exec"
        "strings"
        "sync"

        grpcprom "github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus"
        "google.golang.org/grpc"
        "k8s.io/klog/v2"
        "sigs.k8s.io/blob-csi-driver/pkg/blob"
        mount_azure_blob "sigs.k8s.io/blob-csi-driver/pkg/blobfuse-proxy/pb"
        "sigs.k8s.io/blob-csi-driver/pkg/util"
)

var (
        mutex sync.Mutex
)

type BlobfuseVersion int

const (
        BlobfuseV1 BlobfuseVersion = iota
        BlobfuseV2
)

type MountServer struct {
        blobfuseVersion BlobfuseVersion
        mount_azure_blob.UnimplementedMountServiceServer
}

// NewMountServer returns a new Mountserver
func NewMountServiceServer() *MountServer {
        mountServer := &MountServer{}
        mountServer.blobfuseVersion = getBlobfuseVersion()
        return mountServer
}

// MountAzureBlob mounts an azure blob container to given location
func (server *MountServer) MountAzureBlob(_ context.Context,
        req *mount_azure_blob.MountAzureBlobRequest,
) (resp *mount_azure_blob.MountAzureBlobResponse, err error) {
        mutex.Lock()
        defer mutex.Unlock()

        args := req.GetMountArgs()
        authEnv := req.GetAuthEnv()
        protocol := req.GetProtocol()
        klog.V(2).Infof("received mount request: protocol: %s, server default blobfuseVersion: %v, mount args %v \n", protocol, server.blobfuseVersion, args)

        var cmd *exec.Cmd
        var result mount_azure_blob.MountAzureBlobResponse
        if protocol == blob.Fuse2 || server.blobfuseVersion == BlobfuseV2 {
                args = "mount " + args
                // add this arg for blobfuse2 to solve the issue:
                // https://github.com/Azure/azure-storage-fuse/issues/1015
                if !strings.Contains(args, "--ignore-open-flags") {
                        klog.V(2).Infof("append --ignore-open-flags=true to mount args")
                        args = args + " " + "--ignore-open-flags=true"
                }
                if !strings.Contains(args, "--disable-version-check") {
                        klog.V(2).Infof("append --disable-version-check to mount args")
                        args = args + " " + "--disable-version-check=true"
                }
                args = util.TrimDuplicatedSpace(args)
                klog.V(2).Infof("mount with v2, protocol: %s, args: %s", protocol, args)
                cmd = exec.Command("blobfuse2", strings.Split(args, " ")...)
        } else {
                args = util.TrimDuplicatedSpace(args)
                klog.V(2).Infof("mount with v1, protocol: %s, args: %s", protocol, args)
                cmd = exec.Command("blobfuse", strings.Split(args, " ")...)
        }

        cmd.Env = append(os.Environ(), authEnv...)
        output, err := cmd.CombinedOutput()
        if err != nil {
                klog.Error("blobfuse mount failed: with error:", err.Error())
        } else {
                klog.V(2).Infof("successfully mounted")
        }
        result.Output = string(output)
        klog.V(2).Infof("blobfuse output: %s\n", result.Output)
        if err != nil {
                return &result, fmt.Errorf("%w %s", err, result.Output)
        }
        return &result, nil
}

func RunGRPCServer(
        mountServer mount_azure_blob.MountServiceServer,
        enableTLS bool,
        listener net.Listener,
) error {
        serverOptions := []grpc.ServerOption{
                grpc.ChainUnaryInterceptor(
                        grpcprom.NewServerMetrics().UnaryServerInterceptor(),
                ),
        }

        grpcServer := grpc.NewServer(serverOptions...)

        mount_azure_blob.RegisterMountServiceServer(grpcServer, mountServer)

        klog.V(2).Infof("Start GRPC server at %s, TLS = %t", listener.Addr().String(), enableTLS)
        return grpcServer.Serve(listener)
}

func getBlobfuseVersion() BlobfuseVersion {
        osinfo, err := util.GetOSInfo("/etc/os-release")
        if err != nil {
                klog.Warningf("failed to get OS info: %v, default using blobfuse v1", err)
                return BlobfuseV1
        }

        if (strings.EqualFold(osinfo.Distro, "mariner") || strings.EqualFold(osinfo.Distro, "azurelinux")) && osinfo.Version >= "2.0" {
                klog.V(2).Info("proxy default using blobfuse V2 for mounting on azurelinux(mariner) 2.0+")
                return BlobfuseV2
        }

        if strings.EqualFold(osinfo.Distro, "rhcos") {
                klog.V(2).Info("proxy default using blobfuse V2 for mounting on RHCOS")
                return BlobfuseV2
        }

        if strings.EqualFold(osinfo.Distro, "ubuntu") && osinfo.Version >= "22.04" {
                klog.V(2).Info("proxy default using blobfuse V2 for mounting on Ubuntu 22.04+")
                return BlobfuseV2
        }

        klog.V(2).Info("proxy default using blobfuse V1 for mounting")
        return BlobfuseV1
}

1	/*
2	Copyright 2021 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 server
18
19	import (
20	"context"
21	"fmt"
22	"net"
23	"os"
24	"os/exec"
25	"strings"
26	"sync"
27
28	grpcprom "github.com/grpc-ecosystem/go-grpc-middleware/providers/prometheus"
29	"google.golang.org/grpc"
30	"k8s.io/klog/v2"
31	"sigs.k8s.io/blob-csi-driver/pkg/blob"
32	mount_azure_blob "sigs.k8s.io/blob-csi-driver/pkg/blobfuse-proxy/pb"
33	"sigs.k8s.io/blob-csi-driver/pkg/util"
34	)
35
36	var (
37	mutex sync.Mutex
38	)
39
40	type BlobfuseVersion int
41
42	const (
43	BlobfuseV1 BlobfuseVersion = iota
44	BlobfuseV2
45	)
46
47	type MountServer struct {
48	blobfuseVersion BlobfuseVersion
49	mount_azure_blob.UnimplementedMountServiceServer
50	}
51
52	// NewMountServer returns a new Mountserver
53	func NewMountServiceServer() *MountServer {	1✔
54	mountServer := &MountServer{}	1✔
55	mountServer.blobfuseVersion = getBlobfuseVersion()	1✔
56	return mountServer	1✔
57	}	1✔
58
59	// MountAzureBlob mounts an azure blob container to given location
60	func (server *MountServer) MountAzureBlob(_ context.Context,
61	req *mount_azure_blob.MountAzureBlobRequest,
62	) (resp *mount_azure_blob.MountAzureBlobResponse, err error) {	1✔
63	mutex.Lock()	1✔
64	defer mutex.Unlock()	1✔
65		1✔
66	args := req.GetMountArgs()	1✔
67	authEnv := req.GetAuthEnv()	1✔
68	protocol := req.GetProtocol()	1✔
69	klog.V(2).Infof("received mount request: protocol: %s, server default blobfuseVersion: %v, mount args %v \n", protocol, server.blobfuseVersion, args)	1✔
70		1✔
71	var cmd *exec.Cmd	1✔
72	var result mount_azure_blob.MountAzureBlobResponse	1✔
73	if protocol == blob.Fuse2 \|\| server.blobfuseVersion == BlobfuseV2 {	2✔
74	args = "mount " + args	1✔
75	// add this arg for blobfuse2 to solve the issue:	1✔
76	// https://github.com/Azure/azure-storage-fuse/issues/1015	1✔
77	if !strings.Contains(args, "--ignore-open-flags") {	2✔
78	klog.V(2).Infof("append --ignore-open-flags=true to mount args")	1✔
79	args = args + " " + "--ignore-open-flags=true"	1✔
80	}	1✔
81	if !strings.Contains(args, "--disable-version-check") {	2✔
82	klog.V(2).Infof("append --disable-version-check to mount args")	1✔
83	args = args + " " + "--disable-version-check=true"	1✔
84	}	1✔
85	args = util.TrimDuplicatedSpace(args)	1✔
86	klog.V(2).Infof("mount with v2, protocol: %s, args: %s", protocol, args)	1✔
87	cmd = exec.Command("blobfuse2", strings.Split(args, " ")...)	1✔
88	} else {	×
89	args = util.TrimDuplicatedSpace(args)	×
90	klog.V(2).Infof("mount with v1, protocol: %s, args: %s", protocol, args)	×
91	cmd = exec.Command("blobfuse", strings.Split(args, " ")...)	×
92	}	×
93
94	cmd.Env = append(os.Environ(), authEnv...)	1✔
95	output, err := cmd.CombinedOutput()	1✔
96	if err != nil {	2✔
97	klog.Error("blobfuse mount failed: with error:", err.Error())	1✔
98	} else {	1✔
99	klog.V(2).Infof("successfully mounted")	×
100	}	×
101	result.Output = string(output)	1✔
102	klog.V(2).Infof("blobfuse output: %s\n", result.Output)	1✔
103	if err != nil {	2✔
104	return &result, fmt.Errorf("%w %s", err, result.Output)	1✔
105	}	1✔
106	return &result, nil	×
107	}
108
109	func RunGRPCServer(
110	mountServer mount_azure_blob.MountServiceServer,
111	enableTLS bool,
112	listener net.Listener,
113	) error {	×
114	serverOptions := []grpc.ServerOption{	×
115	grpc.ChainUnaryInterceptor(	×
116	grpcprom.NewServerMetrics().UnaryServerInterceptor(),	×
117	),	×
118	}	×
119		×
120	grpcServer := grpc.NewServer(serverOptions...)	×
121		×
122	mount_azure_blob.RegisterMountServiceServer(grpcServer, mountServer)	×
123		×
124	klog.V(2).Infof("Start GRPC server at %s, TLS = %t", listener.Addr().String(), enableTLS)	×
125	return grpcServer.Serve(listener)	×
126	}	×
127
128	func getBlobfuseVersion() BlobfuseVersion {	1✔
129	osinfo, err := util.GetOSInfo("/etc/os-release")	1✔
130	if err != nil {	1✔
131	klog.Warningf("failed to get OS info: %v, default using blobfuse v1", err)	×
132	return BlobfuseV1	×
133	}	×
134
135	if (strings.EqualFold(osinfo.Distro, "mariner") \|\| strings.EqualFold(osinfo.Distro, "azurelinux")) && osinfo.Version >= "2.0" {	1✔
136	klog.V(2).Info("proxy default using blobfuse V2 for mounting on azurelinux(mariner) 2.0+")	×
137	return BlobfuseV2	×
138	}	×
139
140	if strings.EqualFold(osinfo.Distro, "rhcos") {	1✔
141	klog.V(2).Info("proxy default using blobfuse V2 for mounting on RHCOS")	×
142	return BlobfuseV2	×
143	}	×
144
145	if strings.EqualFold(osinfo.Distro, "ubuntu") && osinfo.Version >= "22.04" {	2✔
146	klog.V(2).Info("proxy default using blobfuse V2 for mounting on Ubuntu 22.04+")	1✔
147	return BlobfuseV2	1✔
148	}	1✔
149
150	klog.V(2).Info("proxy default using blobfuse V1 for mounting")	×
151	return BlobfuseV1	×
152	}

kubernetes-sigs / blob-csi-driver / 16064476868

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