From ca8a32f0e995d53d8e5b64e82812cdb26cfe26d6 Mon Sep 17 00:00:00 2001 From: Amit Roushan Date: Wed, 18 Aug 2021 19:06:29 +0530 Subject: [PATCH] add topology awareness for node plugin --- src/csi/driver/driver.go | 12 ++- src/csi/driver/identity.go | 7 ++ src/csi/driver/node.go | 18 ++++- src/csi/main.go | 24 +++++- src/utils/k8sutils/k8sutils.go | 75 +++++++++++++++++++ yamls/deploy/huawei-csi-controller.yaml | 5 ++ yamls/deploy/huawei-csi-multi-controller.yaml | 5 ++ yamls/deploy/huawei-csi-node.yaml | 6 ++ yamls/deploy/huawei-csi-rbac.yaml | 3 + .../deploy/huawei-csi-resize-controller.yaml | 5 ++ yamls/deploy/huawei-csi-resize-rbac.yaml | 3 + ...huawei-csi-resize-snapshot-controller.yaml | 5 ++ .../huawei-csi-resize-snapshot-rbac.yaml | 3 + 13 files changed, 166 insertions(+), 5 deletions(-) create mode 100644 src/utils/k8sutils/k8sutils.go diff --git a/src/csi/driver/driver.go b/src/csi/driver/driver.go index 89e96804..732a8446 100644 --- a/src/csi/driver/driver.go +++ b/src/csi/driver/driver.go @@ -1,17 +1,27 @@ package driver +import ( + "strings" + "utils/k8sutils" +) + type Driver struct { name string version string useMultiPath bool isNeedMultiPath bool + k8sUtils k8sutils.Interface + nodeName string } -func NewDriver(name, version string, useMultiPath, isNeedMultiPath bool) *Driver { +func NewDriver(name, version string, useMultiPath, isNeedMultiPath bool, + k8sUtils k8sutils.Interface, nodeName string) *Driver { return &Driver{ name: name, version: version, useMultiPath: useMultiPath, isNeedMultiPath: isNeedMultiPath, + k8sUtils: k8sUtils, + nodeName: strings.TrimSpace(nodeName), } } diff --git a/src/csi/driver/identity.go b/src/csi/driver/identity.go index b6868e09..731c4765 100644 --- a/src/csi/driver/identity.go +++ b/src/csi/driver/identity.go @@ -27,6 +27,13 @@ func (d *Driver) GetPluginCapabilities(ctx context.Context, req *csi.GetPluginCa }, }, }, + &csi.PluginCapability{ + Type: &csi.PluginCapability_Service_{ + Service: &csi.PluginCapability_Service{ + Type: csi.PluginCapability_Service_VOLUME_ACCESSIBILITY_CONSTRAINTS, + }, + }, + }, }, }, nil } diff --git a/src/csi/driver/node.go b/src/csi/driver/node.go index e57976f5..6c717377 100644 --- a/src/csi/driver/node.go +++ b/src/csi/driver/node.go @@ -171,10 +171,26 @@ func (d *Driver) NodeGetInfo(ctx context.Context, req *csi.NodeGetInfoRequest) ( log.Errorf("Marshal node info of %s error: %v", nodeBytes, err) return nil, status.Error(codes.Internal, err.Error()) } - log.Infof("Get NodeId %s", nodeBytes) + + if d.nodeName == "" { + return &csi.NodeGetInfoResponse{ + NodeId: string(nodeBytes), + }, nil + } + + // Get topology info from Node labels + topology, err := d.k8sUtils.GetNodeTopology(d.nodeName) + if err != nil { + log.Errorln(err) + return nil, status.Error(codes.Internal, err.Error()) + } + return &csi.NodeGetInfoResponse{ NodeId: string(nodeBytes), + AccessibleTopology: &csi.Topology{ + Segments: topology, + }, }, nil } diff --git a/src/csi/main.go b/src/csi/main.go index 4cd9a617..5f781d14 100644 --- a/src/csi/main.go +++ b/src/csi/main.go @@ -14,6 +14,7 @@ import ( "runtime/debug" "time" "utils" + "utils/k8sutils" "utils/log" "github.com/container-storage-interface/spec/lib/go/csi" @@ -30,6 +31,8 @@ const ( csiVersion = "2.2.13" defaultDriverName = "csi.huawei.com" + + nodeNameEnv = "CSI_NODENAME" ) var ( @@ -54,6 +57,12 @@ var ( volumeUseMultiPath = flag.Bool("volume-use-multipath", true, "Whether to use multipath when attach block volume") + kubeconfig = flag.String("kubeconfig", + "", + "absolute path to the kubeconfig file") + nodeName = flag.String("nodename", + os.Getenv(nodeNameEnv), + "absolute path to the kubeconfig file") config CSIConfig secret CSISecret @@ -64,7 +73,7 @@ type CSIConfig struct { } type CSISecret struct { - Secrets map[string]interface{} `json:"secrets"` + Secrets map[string]interface{} `json:"secrets"` } func init() { @@ -97,6 +106,10 @@ func init() { _ = mergeData(config, secret) + if "" == *nodeName { + logrus.Warning("Node name is empty. Topology aware volume provisioning feature may not behave normal") + } + if *containerized { *controllerFlagFile = "" } @@ -199,10 +212,15 @@ func main() { log.Fatalf("Listen on %s error: %v", *endpoint, err) } + k8sUtils, err := k8sutils.NewK8SUtils(*kubeconfig) + if err != nil { + log.Fatalf("Kubernetes client initialization failed %v", err) + } + isNeedMultiPath := utils.NeedMultiPath(config.Backends) - d := driver.NewDriver(*driverName, csiVersion, *volumeUseMultiPath, isNeedMultiPath) - server := grpc.NewServer() + d := driver.NewDriver(*driverName, csiVersion, *volumeUseMultiPath, isNeedMultiPath, k8sUtils, *nodeName) + server := grpc.NewServer() csi.RegisterIdentityServer(server, d) csi.RegisterControllerServer(server, d) csi.RegisterNodeServer(server, d) diff --git a/src/utils/k8sutils/k8sutils.go b/src/utils/k8sutils/k8sutils.go new file mode 100644 index 00000000..dff1deac --- /dev/null +++ b/src/utils/k8sutils/k8sutils.go @@ -0,0 +1,75 @@ +package k8sutils + +import ( + "errors" + "fmt" + "regexp" + + corev1 "k8s.io/api/core/v1" + metav1 "k8s.io/apimachinery/pkg/apis/meta/v1" + "k8s.io/client-go/kubernetes" + "k8s.io/client-go/rest" + "k8s.io/client-go/tools/clientcmd" +) + +const ( + topologyRegx = "topology.kubernetes.io/*" +) + +type Interface interface { + GetNodeTopology(nodeName string) (map[string]string, error) +} + +type KubeClient struct { + clientSet *kubernetes.Clientset +} + +func NewK8SUtils(kubeConfig string) (Interface, error) { + var clientset *kubernetes.Clientset + + if kubeConfig != "" { + config, err := clientcmd.BuildConfigFromFlags("", kubeConfig) + if err != nil { + return nil, err + } + + clientset, err = kubernetes.NewForConfig(config) + if err != nil { + return nil, err + } + } else { + config, err := rest.InClusterConfig() + if err != nil { + return nil, err + } + + clientset, err = kubernetes.NewForConfig(config) + if err != nil { + return nil, err + } + } + + return &KubeClient{ + clientSet: clientset, + }, nil +} + +func (k *KubeClient) GetNodeTopology(nodeName string) (map[string]string, error) { + k8sNode, err := k.getNode(nodeName) + if err != nil { + return nil, errors.New(fmt.Sprintf("failed to get node topology with error: %v", err)) + } + + topology := make(map[string]string) + for key, value := range k8sNode.Labels { + if match, err := regexp.MatchString(topologyRegx, key); err == nil && match { + topology[key] = value + } + } + + return topology, nil +} + +func (k *KubeClient) getNode(nodeName string) (*corev1.Node, error) { + return k.clientSet.CoreV1().Nodes().Get(nodeName, metav1.GetOptions{}) +} diff --git a/yamls/deploy/huawei-csi-controller.yaml b/yamls/deploy/huawei-csi-controller.yaml index ca43b9bb..069d553c 100644 --- a/yamls/deploy/huawei-csi-controller.yaml +++ b/yamls/deploy/huawei-csi-controller.yaml @@ -52,6 +52,11 @@ spec: env: - name: CSI_ENDPOINT value: /var/lib/csi/sockets/pluginproxy/csi.sock + - name: CSI_NODENAME + valueFrom: + fieldRef: + apiVersion: v1 + fieldPath: spec.nodeName imagePullPolicy: "IfNotPresent" volumeMounts: - name: socket-dir diff --git a/yamls/deploy/huawei-csi-multi-controller.yaml b/yamls/deploy/huawei-csi-multi-controller.yaml index b21753d7..1a988278 100644 --- a/yamls/deploy/huawei-csi-multi-controller.yaml +++ b/yamls/deploy/huawei-csi-multi-controller.yaml @@ -55,6 +55,11 @@ spec: env: - name: CSI_ENDPOINT value: /var/lib/csi/sockets/pluginproxy/csi.sock + - name: CSI_NODENAME + valueFrom: + fieldRef: + apiVersion: v1 + fieldPath: spec.nodeName imagePullPolicy: "IfNotPresent" volumeMounts: - name: socket-dir diff --git a/yamls/deploy/huawei-csi-node.yaml b/yamls/deploy/huawei-csi-node.yaml index 760694d4..b612187c 100644 --- a/yamls/deploy/huawei-csi-node.yaml +++ b/yamls/deploy/huawei-csi-node.yaml @@ -36,6 +36,12 @@ spec: - "--containerized" - "--driver-name=csi.huawei.com" - "--volume-use-multipath=true" + env: + - name: CSI_NODENAME + valueFrom: + fieldRef: + apiVersion: v1 + fieldPath: spec.nodeName securityContext: privileged: true capabilities: diff --git a/yamls/deploy/huawei-csi-rbac.yaml b/yamls/deploy/huawei-csi-rbac.yaml index f90380a2..6e81156d 100644 --- a/yamls/deploy/huawei-csi-rbac.yaml +++ b/yamls/deploy/huawei-csi-rbac.yaml @@ -100,6 +100,9 @@ rules: - apiGroups: [""] resources: ["events"] verbs: ["get", "list", "watch", "create", "update", "patch"] + - apiGroups: [""] + resources: ["nodes"] + verbs: ["get"] --- kind: ClusterRoleBinding diff --git a/yamls/deploy/huawei-csi-resize-controller.yaml b/yamls/deploy/huawei-csi-resize-controller.yaml index c2df7794..a3b19afb 100644 --- a/yamls/deploy/huawei-csi-resize-controller.yaml +++ b/yamls/deploy/huawei-csi-resize-controller.yaml @@ -65,6 +65,11 @@ spec: env: - name: CSI_ENDPOINT value: /var/lib/csi/sockets/pluginproxy/csi.sock + - name: CSI_NODENAME + valueFrom: + fieldRef: + apiVersion: v1 + fieldPath: spec.nodeName imagePullPolicy: "IfNotPresent" volumeMounts: - name: socket-dir diff --git a/yamls/deploy/huawei-csi-resize-rbac.yaml b/yamls/deploy/huawei-csi-resize-rbac.yaml index 57e11aee..3ca0f90b 100644 --- a/yamls/deploy/huawei-csi-resize-rbac.yaml +++ b/yamls/deploy/huawei-csi-resize-rbac.yaml @@ -164,6 +164,9 @@ rules: - apiGroups: [""] resources: ["events"] verbs: ["get", "list", "watch", "create", "update", "patch"] + - apiGroups: [""] + resources: ["nodes"] + verbs: ["get"] --- kind: ClusterRoleBinding diff --git a/yamls/deploy/huawei-csi-resize-snapshot-controller.yaml b/yamls/deploy/huawei-csi-resize-snapshot-controller.yaml index c4835271..f78729cb 100644 --- a/yamls/deploy/huawei-csi-resize-snapshot-controller.yaml +++ b/yamls/deploy/huawei-csi-resize-snapshot-controller.yaml @@ -88,6 +88,11 @@ spec: env: - name: CSI_ENDPOINT value: /var/lib/csi/sockets/pluginproxy/csi.sock + - name: CSI_NODENAME + valueFrom: + fieldRef: + apiVersion: v1 + fieldPath: spec.nodeName imagePullPolicy: "IfNotPresent" volumeMounts: - name: socket-dir diff --git a/yamls/deploy/huawei-csi-resize-snapshot-rbac.yaml b/yamls/deploy/huawei-csi-resize-snapshot-rbac.yaml index 0a189402..779bf7a1 100644 --- a/yamls/deploy/huawei-csi-resize-snapshot-rbac.yaml +++ b/yamls/deploy/huawei-csi-resize-snapshot-rbac.yaml @@ -298,6 +298,9 @@ rules: - apiGroups: [""] resources: ["events"] verbs: ["get", "list", "watch", "create", "update", "patch"] + - apiGroups: [""] + resources: ["nodes"] + verbs: ["get"] --- kind: ClusterRoleBinding