4891009596

Committed 05 May 2023 07:39AM UTC coverage: 80.779% (+0.03%) from 80.747%

Build Type

push

github

Committed by GitHub

Commit Message

Merge pull request #909 from andyzhangx/trim_duplicated_space-1.20

Run Details

9 of 9 new or added lines in 3 files covered. (100.0%)

1824 of 2258 relevant lines covered (80.78%)

5.28 hits per line

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

66.67

/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/exec"
        "strings"
        "sync"

        "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(ctx 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, Mounting with args %v \n", protocol, server.blobfuseVersion, args)

        var cmd *exec.Cmd
        var result mount_azure_blob.MountAzureBlobResponse
        if protocol == blob.Fuse2 || server.blobfuseVersion == BlobfuseV2 {
                klog.V(2).Infof("using blobfuse V2 to mount")
                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"
                }
                args = util.TrimDuplicatedSpace(args)
                cmd = exec.Command("blobfuse2", strings.Split(args, " ")...)
        } else {
                args = util.TrimDuplicatedSpace(args)
                klog.V(2).Infof("using blobfuse V1 to mount")
                cmd = exec.Command("blobfuse", strings.Split(args, " ")...)
        }

        cmd.Env = append(cmd.Env, 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{}
        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/lsb-release")
        if err != nil {
                klog.Warningf("failed to get OS info: %v, default using blobfuse v1", err)
                return BlobfuseV1
        }

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

kubernetes-sigs / blob-csi-driver / 4891009596

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