Refactor Fluid CSI Plugin (#1395)

* Refactor Fluid CSI Plugin Signed-off-by: TrafalgarZZZ <[email protected]> * Go fmt Signed-off-by: TrafalgarZZZ <[email protected]> * Option to enable fuse mount point recovery Signed-off-by: TrafalgarZZZ <[email protected]> * Refactor Fluid CSI Plugin - Move package 'mountinfo' to utils - Move package 'fuse' to plugin - Add registry functions Signed-off-by: TrafalgarZZZ <[email protected]> * Minor fix for fuse recover period Signed-off-by: TrafalgarZZZ <[email protected]> * Remove duplicate license Signed-off-by: TrafalgarZZZ <[email protected]> * Declare registraion funcs to enable/disable feature gates Signed-off-by: TrafalgarZZZ <[email protected]> * Add tests Signed-off-by: TrafalgarZZZ <[email protected]> * Add tests Signed-off-by: TrafalgarZZZ <[email protected]> * Fix copyright Signed-off-by: TrafalgarZZZ <[email protected]>
fluid-cloudnative · Jan 26, 2022 · 935578a · 935578a
1 parent 02a5698
commit 935578a
Show file tree

Hide file tree

Showing 14 changed files with 407 additions and 115 deletions.
diff --git a/cmd/csi/app/csi.go b/cmd/csi/app/csi.go
@@ -23,24 +23,16 @@ import (
 	"fmt"
 	"github.com/fluid-cloudnative/fluid"
 	datav1alpha1 "github.com/fluid-cloudnative/fluid/api/v1alpha1"
-	csi "github.com/fluid-cloudnative/fluid/pkg/csi/fuse"
-	"github.com/fluid-cloudnative/fluid/pkg/csi/recover"
-	"github.com/fluid-cloudnative/fluid/pkg/utils"
-	"github.com/fluid-cloudnative/fluid/pkg/utils/kubelet"
+	"github.com/fluid-cloudnative/fluid/pkg/csi"
+	"github.com/fluid-cloudnative/fluid/pkg/csi/config"
 	"github.com/golang/glog"
 	"github.com/spf13/cobra"
-	"io/ioutil"
 	"k8s.io/apimachinery/pkg/runtime"
-	"k8s.io/apimachinery/pkg/util/wait"
 	clientgoscheme "k8s.io/client-go/kubernetes/scheme"
-	"k8s.io/client-go/rest"
-	"k8s.io/client-go/tools/record"
 	"net/http"
 	"net/http/pprof"
 	"os"
 	ctrl "sigs.k8s.io/controller-runtime"
-	"sigs.k8s.io/controller-runtime/pkg/client"
-	"strconv"
 	"time"
 )
 
@@ -52,8 +44,6 @@ var (
 	recoverFusePeriod int
 )
 
-const defaultKubeletTimeout = 10
-
 var scheme = runtime.NewScheme()
 
 var startCmd = &cobra.Command{
@@ -143,74 +133,21 @@ func handle() {
 	})
 
 	if err != nil {
-		panic(fmt.Sprintf("csi: unable to create controller recover due to error %v", err))
+		panic(fmt.Sprintf("csi: unable to create controller manager due to error %v", err))
 	}
 
-	ctx := ctrl.SetupSignalHandler()
-	go func() {
-		if err := mgr.Start(ctx); err != nil {
-			panic(fmt.Sprintf("unable to start controller recover due to error %v", err))
-		}
-	}()
-
-	if recoverFusePeriod > 0 {
-		if err := recoverStart(mgr.GetClient(), mgr.GetEventRecorderFor("FuseRecover")); err != nil {
-			panic(fmt.Sprintf("unable to start recover due to error %v", err))
-		}
+	config := config.Config{
+		NodeId:            nodeID,
+		Endpoint:          endpoint,
+		RecoverFusePeriod: recoverFusePeriod,
 	}
 
-	d := csi.NewDriver(nodeID, endpoint, mgr.GetClient())
-	d.Run()
-}
-
-func recoverStart(kubeClient client.Client, recorder record.EventRecorder) (err error) {
-	glog.V(3).Infoln("start csi recover")
-	mountRoot, err := utils.GetMountRoot()
-	if err != nil {
-		return
+	if err = csi.SetupWithManager(mgr, config); err != nil {
+		panic(fmt.Sprintf("unable to set up manager due to error %v", err))
 	}
-	glog.V(3).Infof("Get mount root: %s", mountRoot)
 
-	// get CSI sa token
-	tokenByte, err := ioutil.ReadFile("/var/run/secrets/kubernetes.io/serviceaccount/token")
-	if err != nil {
-		panic(fmt.Errorf("in cluster mode, find token failed, error: %v", err))
-	}
-	token := string(tokenByte)
-
-	glog.V(3).Infoln("start kubelet client")
-	nodeIp := os.Getenv("NODE_IP")
-	kubeletClientCert := os.Getenv("KUBELET_CLIENT_CERT")
-	kubeletClientKey := os.Getenv("KUBELET_CLIENT_KEY")
-	var kubeletTimeout int
-	if os.Getenv("KUBELET_TIMEOUT") != "" {
-		if kubeletTimeout, err = strconv.Atoi(os.Getenv("KUBELET_TIMEOUT")); err != nil {
-			glog.Errorf("parse kubelet timeout error: %v", err)
-			return
-		}
-	} else {
-		kubeletTimeout = defaultKubeletTimeout
-	}
-	glog.V(3).Infof("get node ip: %s", nodeIp)
-	kubeletClient, err := kubelet.NewKubeletClient(&kubelet.KubeletClientConfig{
-		Address: nodeIp,
-		Port:    10250,
-		TLSClientConfig: rest.TLSClientConfig{
-			ServerName: "kubelet",
-			CertFile:   kubeletClientCert,
-			KeyFile:    kubeletClientKey,
-		},
-		BearerToken: token,
-		HTTPTimeout: time.Duration(kubeletTimeout) * time.Second,
-	})
-	if err != nil {
-		glog.Error(err)
-		return
+	ctx := ctrl.SetupSignalHandler()
+	if err = mgr.Start(ctx); err != nil {
+		panic(fmt.Sprintf("unable to start controller recover due to error %v", err))
 	}
-	m := recover.NewFuseRecoder(kubeClient, kubeletClient, recorder)
-	// do recovering at beginning
-	// recover set containerStat in memory, it's none when start
-	m.Recover()
-	go m.Run(recoverFusePeriod, wait.NeverStop)
-	return
 }
diff --git a/pkg/csi/config/config.go b/pkg/csi/config/config.go
@@ -0,0 +1,23 @@
+/*
+Copyright 2022 The Fluid 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 config
+
+type Config struct {
+	NodeId            string
+	Endpoint          string
+	RecoverFusePeriod int
+}
diff --git a/pkg/csi/fuse/controller.go → pkg/csi/plugins/controller.go b/pkg/csi/fuse/controller.go → pkg/csi/plugins/controller.go
@@ -13,7 +13,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package csi
+package plugins
 
 import (
 	"crypto/sha1"

diff --git a/pkg/csi/fuse/driver.go → pkg/csi/plugins/driver.go b/pkg/csi/fuse/driver.go → pkg/csi/plugins/driver.go
@@ -13,28 +13,15 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-/*
-
-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 csi
+package plugins
 
 import (
+	"context"
 	"fmt"
 	"os"
 	"path/filepath"
 	"sigs.k8s.io/controller-runtime/pkg/client"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
 	"strings"
 
 	"github.com/container-storage-interface/spec/lib/go/csi"
@@ -54,6 +41,8 @@ type driver struct {
 	nodeId, endpoint string
 }
 
+var _ manager.Runnable = &driver{}
+
 func NewDriver(nodeID, endpoint string, client client.Client) *driver {
 	glog.Infof("Driver: %v version: %v", driverName, version)
 
@@ -88,6 +77,7 @@ func (d *driver) newControllerServer() *controllerServer {
 		DefaultControllerServer: csicommon.NewDefaultControllerServer(d.csiDriver),
 	}
 }
+
 func (d *driver) newNodeServer() *nodeServer {
 	return &nodeServer{
 		nodeId:            d.nodeId,
@@ -96,7 +86,7 @@ func (d *driver) newNodeServer() *nodeServer {
 	}
 }
 
-func (d *driver) Run() {
+func (d *driver) run() {
 	s := csicommon.NewNonBlockingGRPCServer()
 	s.Start(
 		d.endpoint,
@@ -106,3 +96,8 @@ func (d *driver) Run() {
 	)
 	s.Wait()
 }
+
+func (d *driver) Start(ctx context.Context) error {
+	d.run()
+	return nil
+}
diff --git a/pkg/csi/fuse/nodeserver.go → pkg/csi/plugins/nodeserver.go b/pkg/csi/fuse/nodeserver.go → pkg/csi/plugins/nodeserver.go
@@ -13,7 +13,7 @@ See the License for the specific language governing permissions and
 limitations under the License.
 */
 
-package csi
+package plugins
 
 import (
 	"fmt"

diff --git a/pkg/csi/plugins/register.go b/pkg/csi/plugins/register.go
@@ -0,0 +1,38 @@
+/*
+Copyright 2022 The Fluid 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 plugins
+
+import (
+	"github.com/fluid-cloudnative/fluid/pkg/csi/config"
+	"sigs.k8s.io/controller-runtime/pkg/manager"
+)
+
+// Register initializes the csi driver and registers it to the controller manager.
+func Register(mgr manager.Manager, cfg config.Config) error {
+	csiDriver := NewDriver(cfg.NodeId, cfg.Endpoint, mgr.GetClient())
+
+	if err := mgr.Add(csiDriver); err != nil {
+		return err
+	}
+
+	return nil
+}
+
+// Enabled checks if the csi driver should be enabled.
+func Enabled(cfg config.Config) bool {
+	return true
+}