Implemented logic in kubelet for registering node info, including wiring to…

hack/.golint_failures

@@ -407,7 +407,6 @@ pkg/volume/azure_file
 pkg/volume/cephfs
 pkg/volume/configmap
 pkg/volume/csi/fake
-pkg/volume/csi/nodeupdater
 pkg/volume/empty_dir
 pkg/volume/fc
 pkg/volume/flexvolume

pkg/volume/csi/BUILD

@@ -16,7 +16,7 @@ go_library(
         "//pkg/features:go_default_library",
         "//pkg/util/strings:go_default_library",
         "//pkg/volume:go_default_library",
-        "//pkg/volume/csi/nodeupdater:go_default_library",
+        "//pkg/volume/csi/nodeinfomanager:go_default_library",
         "//pkg/volume/util:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",
         "//staging/src/k8s.io/api/storage/v1beta1:go_default_library",
@@ -85,7 +85,7 @@ filegroup(
     srcs = [
         ":package-srcs",
         "//pkg/volume/csi/fake:all-srcs",
-        "//pkg/volume/csi/nodeupdater:all-srcs",
+        "//pkg/volume/csi/nodeinfomanager:all-srcs",
     ],
     tags = ["automanaged"],
     visibility = ["//visibility:public"],

pkg/volume/csi/csi_plugin.go

@@ -41,7 +41,7 @@ import (
 	csilister "k8s.io/csi-api/pkg/client/listers/csi/v1alpha1"
 	"k8s.io/kubernetes/pkg/features"
 	"k8s.io/kubernetes/pkg/volume"
-	"k8s.io/kubernetes/pkg/volume/csi/nodeupdater"
+	"k8s.io/kubernetes/pkg/volume/csi/nodeinfomanager"
 )
 
 const (
@@ -99,7 +99,7 @@ type RegistrationHandler struct {
 // corresponding sockets
 var csiDrivers csiDriversStore
 
-var nodeUpdater nodeupdater.Interface
+var nim nodeinfomanager.Interface
 
 // PluginHandler is the plugin registration handler interface passed to the
 // pluginwatcher module in kubelet
@@ -118,12 +118,17 @@ func (h *RegistrationHandler) ValidatePlugin(pluginName string, endpoint string,
 func (h *RegistrationHandler) RegisterPlugin(pluginName string, endpoint string) error {
 	glog.Infof(log("Register new plugin with name: %s at endpoint: %s", pluginName, endpoint))
 
-	// Storing endpoint of newly registered CSI driver into the map, where CSI driver name will be the key
-	// all other CSI components will be able to get the actual socket of CSI drivers by its name.
-	csiDrivers.Lock()
-	defer csiDrivers.Unlock()
+	func() {
+		// Storing endpoint of newly registered CSI driver into the map, where CSI driver name will be the key
+		// all other CSI components will be able to get the actual socket of CSI drivers by its name.
 
-	csiDrivers.driversMap[pluginName] = csiDriver{driverName: pluginName, driverEndpoint: endpoint}
+		// It's not necessary to lock the entire RegistrationCallback() function because only the CSI
+		// client depends on this driver map, and the CSI client does not depend on node information
+		// updated in the rest of the function.
+		csiDrivers.Lock()
+		defer csiDrivers.Unlock()
+		csiDrivers.driversMap[pluginName] = csiDriver{driverName: pluginName, driverEndpoint: endpoint}
+	}()
 
 	// Get node info from the driver.
 	csi := newCsiDriverClient(pluginName)
@@ -131,17 +136,18 @@ func (h *RegistrationHandler) RegisterPlugin(pluginName string, endpoint string)
 	ctx, cancel := context.WithTimeout(context.Background(), csiTimeout)
 	defer cancel()
 
-	driverNodeID, maxVolumePerNode, _, err := csi.NodeGetInfo(ctx)
+	driverNodeID, maxVolumePerNode, accessibleTopology, err := csi.NodeGetInfo(ctx)
 	if err != nil {
+		unregisterDriver(pluginName)
 		return fmt.Errorf("error during CSI NodeGetInfo() call: %v", err)
 	}
 
 	// Calling nodeLabelManager to update annotations and labels for newly registered CSI driver
-	err = nodeUpdater.AddLabelsAndLimits(pluginName, driverNodeID, maxVolumePerNode)
+	// err = nodeUpdater.AddLabelsAndLimits(pluginName, driverNodeID, maxVolumePerNode) // TODO (verult) merge
+	err = nim.AddNodeInfo(pluginName, driverNodeID, maxVolumePerNode, accessibleTopology)
 	if err != nil {
-		// Unregister the driver and return error
-		delete(csiDrivers.driversMap, pluginName)
-		return fmt.Errorf("error while adding CSI labels: %v", err)
+		unregisterDriver(pluginName)
+		return fmt.Errorf("error updating CSI node info in the cluster: %v", err)
 	}
 
 	return nil
@@ -154,12 +160,11 @@ func (h *RegistrationHandler) DeRegisterPlugin(pluginName string) {
 }
 
 func (p *csiPlugin) Init(host volume.VolumeHost) error {
-	glog.Info(log("plugin initializing..."))
 	p.host = host
 
 	// Initializing csiDrivers map and label management channels
 	csiDrivers = csiDriversStore{driversMap: map[string]csiDriver{}}
-	nodeUpdater = nodeupdater.NewNodeUpdater(host.GetNodeName(), host.GetKubeClient())
+	nim = nodeinfomanager.NewNodeInfoManager(host.GetNodeName(), host.GetKubeClient(), host.GetCSIClient())
 
 	csiClient := host.GetCSIClient()
 	if csiClient != nil {
@@ -554,3 +559,15 @@ func (p *csiPlugin) getPublishVolumeInfo(client clientset.Interface, handle, dri
 	}
 	return attachment.Status.AttachmentMetadata, nil
 }
+
+func unregisterDriver(driverName string) {
+	func() {
+		csiDrivers.Lock()
+		defer csiDrivers.Unlock()
+		delete(csiDrivers.driversMap, driverName)
+	}()
+
+	if err := nim.RemoveNodeInfo(driverName); err != nil {
+		glog.Errorf("Error unregistering CSI driver: %v", err)
+	}
+}

pkg/volume/csi/nodeupdater/BUILD → pkg/volume/csi/nodeinfomanager/BUILD

-load("@io_bazel_rules_go//go:def.bzl", "go_library")
+load("@io_bazel_rules_go//go:def.bzl", "go_library", "go_test")
 
 go_library(
     name = "go_default_library",
-    srcs = ["nodeupdater.go"],
-    importpath = "k8s.io/kubernetes/pkg/volume/csi/nodeupdater",
+    srcs = ["nodeinfomanager.go"],
+    importpath = "k8s.io/kubernetes/pkg/volume/csi/nodeinfomanager",
     visibility = ["//visibility:public"],
     deps = [
+        "//pkg/features:go_default_library",
         "//pkg/volume/util:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/api/errors:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/api/resource:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/util/sets:go_default_library",
+        "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
         "//staging/src/k8s.io/client-go/kubernetes:go_default_library",
-        "//staging/src/k8s.io/client-go/kubernetes/typed/core/v1:go_default_library",
         "//staging/src/k8s.io/client-go/util/retry:go_default_library",
+        "//staging/src/k8s.io/csi-api/pkg/apis/csi/v1alpha1:go_default_library",
+        "//staging/src/k8s.io/csi-api/pkg/client/clientset/versioned:go_default_library",
+        "//vendor/github.com/container-storage-interface/spec/lib/go/csi/v0:go_default_library",
         "//vendor/github.com/golang/glog:go_default_library",
     ],
 )
@@ -31,3 +37,21 @@ filegroup(
     tags = ["automanaged"],
     visibility = ["//visibility:public"],
 )
+
+go_test(
+    name = "go_default_test",
+    srcs = ["nodeinfomanager_test.go"],
+    embed = [":go_default_library"],
+    deps = [
+        "//pkg/apis/core/helper:go_default_library",
+        "//staging/src/k8s.io/api/core/v1:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/api/errors:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/util/sets:go_default_library",
+        "//staging/src/k8s.io/client-go/kubernetes/fake:go_default_library",
+        "//staging/src/k8s.io/csi-api/pkg/apis/csi/v1alpha1:go_default_library",
+        "//staging/src/k8s.io/csi-api/pkg/client/clientset/versioned/fake:go_default_library",
+        "//vendor/github.com/container-storage-interface/spec/lib/go/csi/v0:go_default_library",
+    ],
+)

pkg/volume/csi/nodeupdater/nodeupdater.go

-/*
-Copyright 2018 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 nodeupdater includes internal functions used to add/delete labels to
-// kubernetes nodes for corresponding CSI drivers
-package nodeupdater // import "k8s.io/kubernetes/pkg/volume/csi/nodeupdater"
-
-import (
-	"encoding/json"
-	"fmt"
-
-	"github.com/golang/glog"
-	"k8s.io/api/core/v1"
-	"k8s.io/apimachinery/pkg/api/resource"
-	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
-	"k8s.io/apimachinery/pkg/types"
-	"k8s.io/client-go/kubernetes"
-	corev1 "k8s.io/client-go/kubernetes/typed/core/v1"
-	"k8s.io/client-go/util/retry"
-	"k8s.io/kubernetes/pkg/volume/util"
-)
-
-const (
-	// Name of node annotation that contains JSON map of driver names to node
-	// names
-	annotationKey = "csi.volume.kubernetes.io/nodeid"
-)
-
-// labelManagementStruct is struct of channels used for communication between the driver registration
-// code and the go routine responsible for managing the node's labels
-type nodeUpdateStruct struct {
-	nodeName types.NodeName
-	k8s      kubernetes.Interface
-}
-
-// Interface implements an interface for managing labels of a node
-type Interface interface {
-	AddLabelsAndLimits(driverName string, driverNodeId string, maxLimit int64) error
-}
-
-// NewNodeupdater initializes nodeUpdateStruct and returns available interfaces
-func NewNodeUpdater(nodeName types.NodeName, kubeClient kubernetes.Interface) Interface {
-	return nodeUpdateStruct{
-		nodeName: nodeName,
-		k8s:      kubeClient,
-	}
-}
-
-// AddLabelsAndLimits nodeUpdater waits for labeling requests initiated by the driver's registration
-// process and updates labels and attach limits
-func (nodeUpdater nodeUpdateStruct) AddLabelsAndLimits(driverName string, driverNodeId string, maxLimit int64) error {
-	err := addLabelsAndLimits(string(nodeUpdater.nodeName), nodeUpdater.k8s.CoreV1().Nodes(), driverName, driverNodeId, maxLimit)
-	if err != nil {
-		return err
-	}
-	return nil
-}
-
-func addMaxAttachLimitToNode(node *v1.Node, driverName string, maxLimit int64) *v1.Node {
-	if maxLimit <= 0 {
-		glog.V(4).Infof("skipping adding attach limit for %s", driverName)
-		return node
-	}
-
-	if node.Status.Capacity == nil {
-		node.Status.Capacity = v1.ResourceList{}
-	}
-	if node.Status.Allocatable == nil {
-		node.Status.Allocatable = v1.ResourceList{}
-	}
-	limitKeyName := util.GetCSIAttachLimitKey(driverName)
-	node.Status.Capacity[v1.ResourceName(limitKeyName)] = *resource.NewQuantity(maxLimit, resource.DecimalSI)
-	node.Status.Allocatable[v1.ResourceName(limitKeyName)] = *resource.NewQuantity(maxLimit, resource.DecimalSI)
-	return node
-}
-
-// Clones the given map and returns a new map with the given key and value added.
-// Returns the given map, if annotationKey is empty.
-func cloneAndAddAnnotation(
-	annotations map[string]string,
-	annotationKey,
-	annotationValue string) map[string]string {
-	if annotationKey == "" {
-		// Don't need to add an annotation.
-		return annotations
-	}
-	// Clone.
-	newAnnotations := map[string]string{}
-	for key, value := range annotations {
-		newAnnotations[key] = value
-	}
-	newAnnotations[annotationKey] = annotationValue
-	return newAnnotations
-}
-
-func addNodeIdToNode(node *v1.Node, driverName string, csiDriverNodeId string) (*v1.Node, error) {
-	var previousAnnotationValue string
-	if node.ObjectMeta.Annotations != nil {
-		previousAnnotationValue =
-			node.ObjectMeta.Annotations[annotationKey]
-		glog.V(3).Infof(
-			"previousAnnotationValue=%q", previousAnnotationValue)
-	}
-
-	existingDriverMap := map[string]string{}
-	if previousAnnotationValue != "" {
-		// Parse previousAnnotationValue as JSON
-		if err := json.Unmarshal([]byte(previousAnnotationValue), &existingDriverMap); err != nil {
-			return node, fmt.Errorf(
-				"failed to parse node's %q annotation value (%q) err=%v",
-				annotationKey,
-				previousAnnotationValue,
-				err)
-		}
-	}
-
-	if val, ok := existingDriverMap[driverName]; ok {
-		if val == csiDriverNodeId {
-			// Value already exists in node annotation, nothing more to do
-			glog.V(2).Infof(
-				"The key value {%q: %q} alredy eixst in node %q annotation, no need to update: %v",
-				driverName,
-				csiDriverNodeId,
-				annotationKey,
-				previousAnnotationValue)
-			return node, nil
-		}
-	}
-
-	// Add/update annotation value
-	existingDriverMap[driverName] = csiDriverNodeId
-	jsonObj, err := json.Marshal(existingDriverMap)
-	if err != nil {
-		return node, fmt.Errorf(
-			"failed while trying to add key value {%q: %q} to node %q annotation. Existing value: %v",
-			driverName,
-			csiDriverNodeId,
-			annotationKey,
-			previousAnnotationValue)
-	}
-
-	node.ObjectMeta.Annotations = cloneAndAddAnnotation(
-		node.ObjectMeta.Annotations,
-		annotationKey,
-		string(jsonObj))
-	return node, nil
-}
-
-func addLabelsAndLimits(nodeName string, nodeClient corev1.NodeInterface, driverName string, csiDriverNodeId string, maxLimit int64) error {
-	retryErr := retry.RetryOnConflict(retry.DefaultRetry, func() error {
-		// Retrieve the latest version of Node before attempting update, so that
-		// existing changes are not overwritten. RetryOnConflict uses
-		// exponential backoff to avoid exhausting the apiserver.
-		node, getErr := nodeClient.Get(nodeName, metav1.GetOptions{})
-		if getErr != nil {
-			glog.Errorf("Failed to get latest version of Node: %v", getErr)
-			return getErr // do not wrap error
-		}
-		var labelErr error
-		node, labelErr = addNodeIdToNode(node, driverName, csiDriverNodeId)
-		if labelErr != nil {
-			return labelErr
-		}
-		node = addMaxAttachLimitToNode(node, driverName, maxLimit)
-
-		_, updateErr := nodeClient.Update(node)
-		if updateErr == nil {
-			glog.V(2).Infof(
-				"Updated node %q successfully for CSI driver %q and CSI node name %q",
-				nodeName,
-				driverName,
-				csiDriverNodeId)
-		}
-		return updateErr // do not wrap error
-	})
-	if retryErr != nil {
-		return fmt.Errorf("error setting attach limit and labels for %s with : %v", driverName, retryErr)
-	}
-	return nil
-}

plugin/pkg/admission/noderestriction/BUILD

@@ -20,12 +20,14 @@ go_library(
         "//pkg/features:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/api/equality:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/api/errors:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/api/meta:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/util/diff:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/admission:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/admission/initializer:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
         "//staging/src/k8s.io/client-go/informers:go_default_library",
         "//staging/src/k8s.io/client-go/listers/core/v1:go_default_library",
+        "//staging/src/k8s.io/csi-api/pkg/apis/csi/v1alpha1:go_default_library",
     ],
 )
 
@@ -42,12 +44,14 @@ go_test(
         "//pkg/features:go_default_library",
         "//staging/src/k8s.io/api/core/v1:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/apis/meta/v1:go_default_library",
+        "//staging/src/k8s.io/apimachinery/pkg/runtime/schema:go_default_library",
         "//staging/src/k8s.io/apimachinery/pkg/types:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/admission:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/authentication/user:go_default_library",
         "//staging/src/k8s.io/apiserver/pkg/util/feature:go_default_library",
         "//staging/src/k8s.io/client-go/listers/core/v1:go_default_library",
         "//staging/src/k8s.io/client-go/tools/cache:go_default_library",
+        "//staging/src/k8s.io/csi-api/pkg/apis/csi/v1alpha1:go_default_library",
         "//vendor/k8s.io/utils/pointer:go_default_library",
     ],
 )

plugin/pkg/admission/noderestriction/admission.go

@@ -22,12 +22,14 @@ import (
 
 	apiequality "k8s.io/apimachinery/pkg/api/equality"
 	"k8s.io/apimachinery/pkg/api/errors"
+	"k8s.io/apimachinery/pkg/api/meta"
 	"k8s.io/apimachinery/pkg/util/diff"
 	"k8s.io/apiserver/pkg/admission"
 	apiserveradmission "k8s.io/apiserver/pkg/admission/initializer"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	"k8s.io/client-go/informers"
 	corev1lister "k8s.io/client-go/listers/core/v1"
+	csiv1alpha1 "k8s.io/csi-api/pkg/apis/csi/v1alpha1"
 	podutil "k8s.io/kubernetes/pkg/api/pod"
 	authenticationapi "k8s.io/kubernetes/pkg/apis/authentication"
 	coordapi "k8s.io/kubernetes/pkg/apis/coordination"
@@ -87,11 +89,12 @@ func (p *nodePlugin) ValidateInitialization() error {
 }
 
 var (
-	podResource     = api.Resource("pods")
-	nodeResource    = api.Resource("nodes")
-	pvcResource     = api.Resource("persistentvolumeclaims")
-	svcacctResource = api.Resource("serviceaccounts")
-	leaseResource   = coordapi.Resource("leases")
+	podResource         = api.Resource("pods")
+	nodeResource        = api.Resource("nodes")
+	pvcResource         = api.Resource("persistentvolumeclaims")
+	svcacctResource     = api.Resource("serviceaccounts")
+	leaseResource       = coordapi.Resource("leases")
+	csiNodeInfoResource = csiv1alpha1.Resource("csinodeinfos")
 )
 
 func (c *nodePlugin) Admit(a admission.Attributes) error {
@@ -143,6 +146,12 @@ func (c *nodePlugin) Admit(a admission.Attributes) error {
 		}
 		return admission.NewForbidden(a, fmt.Errorf("disabled by feature gate %s", features.NodeLease))
 
+	case csiNodeInfoResource:
+		if c.features.Enabled(features.KubeletPluginsWatcher) {
+			return c.admitCSINodeInfo(nodeName, a)
+		}
+		return admission.NewForbidden(a, fmt.Errorf("disabled by feature gate %s", features.KubeletPluginsWatcher))
+
 	default:
 		return nil
 	}
@@ -422,3 +431,23 @@ func (r *nodePlugin) admitLease(nodeName string, a admission.Attributes) error {
 
 	return nil
 }
+
+func (c *nodePlugin) admitCSINodeInfo(nodeName string, a admission.Attributes) error {
+	// the request must come from a node with the same name as the CSINodeInfo object
+	if a.GetOperation() == admission.Create {
+		// a.GetName() won't return the name on create, so we drill down to the proposed object
+		accessor, err := meta.Accessor(a.GetObject())
+		if err != nil {
+			return admission.NewForbidden(a, fmt.Errorf("unable to access the object name"))
+		}
+		if accessor.GetName() != nodeName {
+			return admission.NewForbidden(a, fmt.Errorf("can only access CSINodeInfo with the same name as the requesting node"))
+		}
+	} else {
+		if a.GetName() != nodeName {
+			return admission.NewForbidden(a, fmt.Errorf("can only access CSINodeInfo with the same name as the requesting node"))
+		}
+	}
+
+	return nil
+}

plugin/pkg/admission/noderestriction/admission_test.go

@@ -23,12 +23,14 @@ import (
 
 	corev1 "k8s.io/api/core/v1"
 	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apimachinery/pkg/types"
 	"k8s.io/apiserver/pkg/admission"
 	"k8s.io/apiserver/pkg/authentication/user"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
 	corev1lister "k8s.io/client-go/listers/core/v1"
 	"k8s.io/client-go/tools/cache"
+	csiv1alpha1 "k8s.io/csi-api/pkg/apis/csi/v1alpha1"
 	authenticationapi "k8s.io/kubernetes/pkg/apis/authentication"
 	"k8s.io/kubernetes/pkg/apis/coordination"
 	api "k8s.io/kubernetes/pkg/apis/core"
@@ -39,10 +41,12 @@ import (
 )
 
 var (
-	trEnabledFeature     = utilfeature.NewFeatureGate()
-	trDisabledFeature    = utilfeature.NewFeatureGate()
-	leaseEnabledFeature  = utilfeature.NewFeatureGate()
-	leaseDisabledFeature = utilfeature.NewFeatureGate()
+	trEnabledFeature              = utilfeature.NewFeatureGate()
+	trDisabledFeature             = utilfeature.NewFeatureGate()
+	leaseEnabledFeature           = utilfeature.NewFeatureGate()
+	leaseDisabledFeature          = utilfeature.NewFeatureGate()
+	pluginsWatcherEnabledFeature  = utilfeature.NewFeatureGate()
+	pluginsWatcherDisabledFeature = utilfeature.NewFeatureGate()
 )
 
 func init() {
@@ -58,6 +62,12 @@ func init() {
 	if err := leaseDisabledFeature.Add(map[utilfeature.Feature]utilfeature.FeatureSpec{features.NodeLease: {Default: false}}); err != nil {
 		panic(err)
 	}
+	if err := pluginsWatcherEnabledFeature.Add(map[utilfeature.Feature]utilfeature.FeatureSpec{features.KubeletPluginsWatcher: {Default: true}}); err != nil {
+		panic(err)
+	}
+	if err := pluginsWatcherDisabledFeature.Add(map[utilfeature.Feature]utilfeature.FeatureSpec{features.KubeletPluginsWatcher: {Default: false}}); err != nil {
+		panic(err)
+	}
 }
 
 func makeTestPod(namespace, name, node string, mirror bool) (*api.Pod, *corev1.Pod) {
@@ -193,6 +203,33 @@ func Test_nodePlugin_Admit(t *testing.T) {
 			},
 		}
 
+		csiNodeInfoResource = csiv1alpha1.Resource("csinodeinfos").WithVersion("v1alpha1")
+		csiNodeInfoKind     = schema.GroupVersionKind{Group: "csi.storage.k8s.io", Version: "v1alpha1", Kind: "CSINodeInfo"}
+		nodeInfo            = &csiv1alpha1.CSINodeInfo{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "mynode",
+			},
+			CSIDrivers: []csiv1alpha1.CSIDriverInfo{
+				{
+					Driver:       "com.example.csi/mydriver",
+					NodeID:       "com.example.csi/mynode",
+					TopologyKeys: []string{"com.example.csi/zone"},
+				},
+			},
+		}
+		nodeInfoWrongName = &csiv1alpha1.CSINodeInfo{
+			ObjectMeta: metav1.ObjectMeta{
+				Name: "foo",
+			},
+			CSIDrivers: []csiv1alpha1.CSIDriverInfo{
+				{
+					Driver:       "com.example.csi/mydriver",
+					NodeID:       "com.example.csi/foo",
+					TopologyKeys: []string{"com.example.csi/zone"},
+				},
+			},
+		}
+
 		noExistingPodsIndex = cache.NewIndexer(cache.MetaNamespaceKeyFunc, nil)
 		noExistingPods      = corev1lister.NewPodLister(noExistingPodsIndex)
 
@@ -955,6 +992,49 @@ func Test_nodePlugin_Admit(t *testing.T) {
 			features:   leaseEnabledFeature,
 			err:        "",
 		},
+		// CSINodeInfo
+		{
+			name:       "disallowed create CSINodeInfo - feature disabled",
+			attributes: admission.NewAttributesRecord(nodeInfo, nil, csiNodeInfoKind, nodeInfo.Namespace, nodeInfo.Name, csiNodeInfoResource, "", admission.Create, false, mynode),
+			features:   pluginsWatcherDisabledFeature,
+			err:        "forbidden: disabled by feature gate KubeletPluginsWatcher",
+		},
+		{
+			name:       "disallowed create another node's CSINodeInfo - feature enabled",
+			attributes: admission.NewAttributesRecord(nodeInfoWrongName, nil, csiNodeInfoKind, nodeInfoWrongName.Namespace, nodeInfoWrongName.Name, csiNodeInfoResource, "", admission.Create, false, mynode),
+			features:   pluginsWatcherEnabledFeature,
+			err:        "forbidden: ",
+		},
+		{
+			name:       "disallowed update another node's CSINodeInfo - feature enabled",
+			attributes: admission.NewAttributesRecord(nodeInfoWrongName, nodeInfoWrongName, csiNodeInfoKind, nodeInfoWrongName.Namespace, nodeInfoWrongName.Name, csiNodeInfoResource, "", admission.Update, false, mynode),
+			features:   pluginsWatcherEnabledFeature,
+			err:        "forbidden: ",
+		},
+		{
+			name:       "disallowed delete another node's CSINodeInfo - feature enabled",
+			attributes: admission.NewAttributesRecord(nil, nil, csiNodeInfoKind, nodeInfoWrongName.Namespace, nodeInfoWrongName.Name, csiNodeInfoResource, "", admission.Delete, false, mynode),
+			features:   pluginsWatcherEnabledFeature,
+			err:        "forbidden: ",
+		},
+		{
+			name:       "allowed create node CSINodeInfo - feature enabled",
+			attributes: admission.NewAttributesRecord(nodeInfo, nil, csiNodeInfoKind, nodeInfo.Namespace, nodeInfo.Name, csiNodeInfoResource, "", admission.Create, false, mynode),
+			features:   pluginsWatcherEnabledFeature,
+			err:        "",
+		},
+		{
+			name:       "allowed update node CSINodeInfo - feature enabled",
+			attributes: admission.NewAttributesRecord(nodeInfo, nodeInfo, csiNodeInfoKind, nodeInfo.Namespace, nodeInfo.Name, csiNodeInfoResource, "", admission.Update, false, mynode),
+			features:   pluginsWatcherEnabledFeature,
+			err:        "",
+		},
+		{
+			name:       "allowed delete node CSINodeInfo - feature enabled",
+			attributes: admission.NewAttributesRecord(nil, nil, csiNodeInfoKind, nodeInfo.Namespace, nodeInfo.Name, csiNodeInfoResource, "", admission.Delete, false, mynode),
+			features:   pluginsWatcherEnabledFeature,
+			err:        "",
+		},
 	}
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {

plugin/pkg/auth/authorizer/node/BUILD

@@ -56,6 +56,7 @@ go_library(
         "//staging/src/k8s.io/client-go/informers/core/v1:go_default_library",
         "//staging/src/k8s.io/client-go/informers/storage/v1beta1:go_default_library",
         "//staging/src/k8s.io/client-go/tools/cache:go_default_library",
+        "//staging/src/k8s.io/csi-api/pkg/apis/csi/v1alpha1:go_default_library",
         "//third_party/forked/gonum/graph:go_default_library",
         "//third_party/forked/gonum/graph/simple:go_default_library",
         "//third_party/forked/gonum/graph/traverse:go_default_library",

plugin/pkg/auth/authorizer/node/node_authorizer.go

@@ -25,6 +25,7 @@ import (
 	"k8s.io/apimachinery/pkg/runtime/schema"
 	"k8s.io/apiserver/pkg/authorization/authorizer"
 	utilfeature "k8s.io/apiserver/pkg/util/feature"
+	csiv1alpha1 "k8s.io/csi-api/pkg/apis/csi/v1alpha1"
 	coordapi "k8s.io/kubernetes/pkg/apis/coordination"
 	api "k8s.io/kubernetes/pkg/apis/core"
 	storageapi "k8s.io/kubernetes/pkg/apis/storage"
@@ -67,13 +68,14 @@ func NewAuthorizer(graph *Graph, identifier nodeidentifier.NodeIdentifier, rules
 }
 
 var (
-	configMapResource = api.Resource("configmaps")
-	secretResource    = api.Resource("secrets")
-	pvcResource       = api.Resource("persistentvolumeclaims")
-	pvResource        = api.Resource("persistentvolumes")
-	vaResource        = storageapi.Resource("volumeattachments")
-	svcAcctResource   = api.Resource("serviceaccounts")
-	leaseResource     = coordapi.Resource("leases")
+	configMapResource   = api.Resource("configmaps")
+	secretResource      = api.Resource("secrets")
+	pvcResource         = api.Resource("persistentvolumeclaims")
+	pvResource          = api.Resource("persistentvolumes")
+	vaResource          = storageapi.Resource("volumeattachments")
+	svcAcctResource     = api.Resource("serviceaccounts")
+	leaseResource       = coordapi.Resource("leases")
+	csiNodeInfoResource = csiv1alpha1.Resource("csinodeinfos")
 )
 
 func (r *NodeAuthorizer) Authorize(attrs authorizer.Attributes) (authorizer.Decision, string, error) {
@@ -120,7 +122,13 @@ func (r *NodeAuthorizer) Authorize(attrs authorizer.Attributes) (authorizer.Deci
 				return r.authorizeLease(nodeName, attrs)
 			}
 			return authorizer.DecisionNoOpinion, fmt.Sprintf("disabled by feature gate %s", features.NodeLease), nil
+		case csiNodeInfoResource:
+			if r.features.Enabled(features.KubeletPluginsWatcher) {
+				return r.authorizeCSINodeInfo(nodeName, attrs)
+			}
+			return authorizer.DecisionNoOpinion, fmt.Sprintf("disabled by feature gate %s", features.KubeletPluginsWatcher), nil
 		}
+
 	}
 
 	// Access to other resources is not subdivided, so just evaluate against the statically defined node rules
@@ -252,6 +260,35 @@ func (r *NodeAuthorizer) authorizeLease(nodeName string, attrs authorizer.Attrib
 	return authorizer.DecisionAllow, "", nil
 }
 
+// authorizeCSINodeInfo authorizes node requests to CSINodeInfo csi.storage.k8s.io/csinodeinfos
+func (r *NodeAuthorizer) authorizeCSINodeInfo(nodeName string, attrs authorizer.Attributes) (authorizer.Decision, string, error) {
+	// allowed verbs: get, create, update, patch, delete
+	verb := attrs.GetVerb()
+	if verb != "get" &&
+		verb != "create" &&
+		verb != "update" &&
+		verb != "patch" &&
+		verb != "delete" {
+		glog.V(2).Infof("NODE DENY: %s %#v", nodeName, attrs)
+		return authorizer.DecisionNoOpinion, "can only get, create, update, patch, or delete a CSINodeInfo", nil
+	}
+
+	if len(attrs.GetSubresource()) > 0 {
+		glog.V(2).Infof("NODE DENY: %s %#v", nodeName, attrs)
+		return authorizer.DecisionNoOpinion, "cannot authorize CSINodeInfo subresources", nil
+	}
+
+	// the request must come from a node with the same name as the CSINodeInfo
+	// note we skip this check for create, since the authorizer doesn't know the name on create
+	// the noderestriction admission plugin is capable of performing this check at create time
+	if verb != "create" && attrs.GetName() != nodeName {
+		glog.V(2).Infof("NODE DENY: %s %#v", nodeName, attrs)
+		return authorizer.DecisionNoOpinion, "can only access CSINodeInfo with the same name as the requesting node", nil
+	}
+
+	return authorizer.DecisionAllow, "", nil
+}
+
 // hasPathFrom returns true if there is a directed path from the specified type/namespace/name to the specified Node
 func (r *NodeAuthorizer) hasPathFrom(nodeName string, startingType vertexType, startingNamespace, startingName string) (bool, error) {
 	r.graph.lock.RLock()

plugin/pkg/auth/authorizer/node/node_authorizer_test.go

@@ -39,12 +39,14 @@ import (
 )
 
 var (
-	csiEnabledFeature    = utilfeature.NewFeatureGate()
-	csiDisabledFeature   = utilfeature.NewFeatureGate()
-	trEnabledFeature     = utilfeature.NewFeatureGate()
-	trDisabledFeature    = utilfeature.NewFeatureGate()
-	leaseEnabledFeature  = utilfeature.NewFeatureGate()
-	leaseDisabledFeature = utilfeature.NewFeatureGate()
+	csiEnabledFeature             = utilfeature.NewFeatureGate()
+	csiDisabledFeature            = utilfeature.NewFeatureGate()
+	trEnabledFeature              = utilfeature.NewFeatureGate()
+	trDisabledFeature             = utilfeature.NewFeatureGate()
+	leaseEnabledFeature           = utilfeature.NewFeatureGate()
+	leaseDisabledFeature          = utilfeature.NewFeatureGate()
+	pluginsWatcherEnabledFeature  = utilfeature.NewFeatureGate()
+	pluginsWatcherDisabledFeature = utilfeature.NewFeatureGate()
 )
 
 func init() {
@@ -66,6 +68,12 @@ func init() {
 	if err := leaseDisabledFeature.Add(map[utilfeature.Feature]utilfeature.FeatureSpec{features.NodeLease: {Default: false}}); err != nil {
 		panic(err)
 	}
+	if err := pluginsWatcherEnabledFeature.Add(map[utilfeature.Feature]utilfeature.FeatureSpec{features.KubeletPluginsWatcher: {Default: true}}); err != nil {
+		panic(err)
+	}
+	if err := pluginsWatcherDisabledFeature.Add(map[utilfeature.Feature]utilfeature.FeatureSpec{features.KubeletPluginsWatcher: {Default: false}}); err != nil {
+		panic(err)
+	}
 }
 
 func TestAuthorizer(t *testing.T) {
@@ -338,6 +346,85 @@ func TestAuthorizer(t *testing.T) {
 			features: leaseEnabledFeature,
 			expect:   authorizer.DecisionAllow,
 		},
+		// CSINodeInfo
+		{
+			name:     "disallowed CSINodeInfo - feature disabled",
+			attrs:    authorizer.AttributesRecord{User: node0, ResourceRequest: true, Verb: "get", Resource: "csinodeinfos", APIGroup: "csi.storage.k8s.io", Name: "node0"},
+			features: pluginsWatcherDisabledFeature,
+			expect:   authorizer.DecisionNoOpinion,
+		},
+		{
+			name:     "disallowed CSINodeInfo with subresource - feature enabled",
+			attrs:    authorizer.AttributesRecord{User: node0, ResourceRequest: true, Verb: "get", Resource: "csinodeinfos", Subresource: "csiDrivers", APIGroup: "csi.storage.k8s.io", Name: "node0"},
+			features: pluginsWatcherEnabledFeature,
+			expect:   authorizer.DecisionNoOpinion,
+		},
+		{
+			name:     "disallowed get another node's CSINodeInfo - feature enabled",
+			attrs:    authorizer.AttributesRecord{User: node0, ResourceRequest: true, Verb: "get", Resource: "csinodeinfos", APIGroup: "csi.storage.k8s.io", Name: "node1"},
+			features: pluginsWatcherEnabledFeature,
+			expect:   authorizer.DecisionNoOpinion,
+		},
+		{
+			name:     "disallowed update another node's CSINodeInfo - feature enabled",
+			attrs:    authorizer.AttributesRecord{User: node0, ResourceRequest: true, Verb: "update", Resource: "csinodeinfos", APIGroup: "csi.storage.k8s.io", Name: "node1"},
+			features: pluginsWatcherEnabledFeature,
+			expect:   authorizer.DecisionNoOpinion,
+		},
+		{
+			name:     "disallowed patch another node's CSINodeInfo - feature enabled",
+			attrs:    authorizer.AttributesRecord{User: node0, ResourceRequest: true, Verb: "patch", Resource: "csinodeinfos", APIGroup: "csi.storage.k8s.io", Name: "node1"},
+			features: pluginsWatcherEnabledFeature,
+			expect:   authorizer.DecisionNoOpinion,
+		},
+		{
+			name:     "disallowed delete another node's CSINodeInfo - feature enabled",
+			attrs:    authorizer.AttributesRecord{User: node0, ResourceRequest: true, Verb: "delete", Resource: "csinodeinfos", APIGroup: "csi.storage.k8s.io", Name: "node1"},
+			features: pluginsWatcherEnabledFeature,
+			expect:   authorizer.DecisionNoOpinion,
+		},
+		{
+			name:     "disallowed list CSINodeInfos - feature enabled",
+			attrs:    authorizer.AttributesRecord{User: node0, ResourceRequest: true, Verb: "list", Resource: "csinodeinfos", APIGroup: "csi.storage.k8s.io"},
+			features: pluginsWatcherEnabledFeature,
+			expect:   authorizer.DecisionNoOpinion,
+		},
+		{
+			name:     "disallowed watch CSINodeInfos - feature enabled",
+			attrs:    authorizer.AttributesRecord{User: node0, ResourceRequest: true, Verb: "watch", Resource: "csinodeinfos", APIGroup: "csi.storage.k8s.io"},
+			features: pluginsWatcherEnabledFeature,
+			expect:   authorizer.DecisionNoOpinion,
+		},
+		{
+			name:     "allowed get CSINodeInfo - feature enabled",
+			attrs:    authorizer.AttributesRecord{User: node0, ResourceRequest: true, Verb: "get", Resource: "csinodeinfos", APIGroup: "csi.storage.k8s.io", Name: "node0"},
+			features: pluginsWatcherEnabledFeature,
+			expect:   authorizer.DecisionAllow,
+		},
+		{
+			name:     "allowed create CSINodeInfo - feature enabled",
+			attrs:    authorizer.AttributesRecord{User: node0, ResourceRequest: true, Verb: "create", Resource: "csinodeinfos", APIGroup: "csi.storage.k8s.io", Name: "node0"},
+			features: pluginsWatcherEnabledFeature,
+			expect:   authorizer.DecisionAllow,
+		},
+		{
+			name:     "allowed update CSINodeInfo - feature enabled",
+			attrs:    authorizer.AttributesRecord{User: node0, ResourceRequest: true, Verb: "update", Resource: "csinodeinfos", APIGroup: "csi.storage.k8s.io", Name: "node0"},
+			features: pluginsWatcherEnabledFeature,
+			expect:   authorizer.DecisionAllow,
+		},
+		{
+			name:     "allowed patch CSINodeInfo - feature enabled",
+			attrs:    authorizer.AttributesRecord{User: node0, ResourceRequest: true, Verb: "patch", Resource: "csinodeinfos", APIGroup: "csi.storage.k8s.io", Name: "node0"},
+			features: pluginsWatcherEnabledFeature,
+			expect:   authorizer.DecisionAllow,
+		},
+		{
+			name:     "allowed delete CSINodeInfo - feature enabled",
+			attrs:    authorizer.AttributesRecord{User: node0, ResourceRequest: true, Verb: "delete", Resource: "csinodeinfos", APIGroup: "csi.storage.k8s.io", Name: "node0"},
+			features: pluginsWatcherEnabledFeature,
+			expect:   authorizer.DecisionAllow,
+		},
 	}
 
 	for _, tc := range tests {

plugin/pkg/auth/authorizer/rbac/bootstrappolicy/policy.go

@@ -164,6 +164,10 @@ func NodeRules() []rbacv1.PolicyRule {
 			nodePolicyRules = append(nodePolicyRules, csiDriverRule)
 		}
 	}
+	if utilfeature.DefaultFeatureGate.Enabled(features.KubeletPluginsWatcher) {
+		csiNodeInfoRule := rbacv1helpers.NewRule("get", "create", "update", "patch", "delete").Groups("csi.storage.k8s.io").Resources("csinodeinfos").RuleOrDie()
+		nodePolicyRules = append(nodePolicyRules, csiNodeInfoRule)
+	}
 
 	// Node leases
 	if utilfeature.DefaultFeatureGate.Enabled(features.NodeLease) {
@@ -453,16 +457,6 @@ func ClusterRoles() []rbacv1.ClusterRole {
 			},
 		},
 		{
-			// a role for the csi external provisioner
-			ObjectMeta: metav1.ObjectMeta{Name: "system:csi-external-provisioner"},
-			Rules: []rbacv1.PolicyRule{
-				rbacv1helpers.NewRule("create", "delete", "get", "list", "watch").Groups(legacyGroup).Resources("persistentvolumes").RuleOrDie(),
-				rbacv1helpers.NewRule("get", "list", "watch", "update", "patch").Groups(legacyGroup).Resources("persistentvolumeclaims").RuleOrDie(),
-				rbacv1helpers.NewRule("list", "watch").Groups(storageGroup).Resources("storageclasses").RuleOrDie(),
-				rbacv1helpers.NewRule("get", "list", "watch", "create", "update", "patch").Groups(legacyGroup).Resources("events").RuleOrDie(),
-			},
-		},
-		{
 			// a role for the csi external attacher
 			ObjectMeta: metav1.ObjectMeta{Name: "system:csi-external-attacher"},
 			Rules: []rbacv1.PolicyRule{
@@ -506,6 +500,22 @@ func ClusterRoles() []rbacv1.ClusterRole {
 		})
 	}
 
+	externalProvisionerRules := []rbacv1.PolicyRule{
+		rbacv1helpers.NewRule("create", "delete", "get", "list", "watch").Groups(legacyGroup).Resources("persistentvolumes").RuleOrDie(),
+		rbacv1helpers.NewRule("get", "list", "watch", "update", "patch").Groups(legacyGroup).Resources("persistentvolumeclaims").RuleOrDie(),
+		rbacv1helpers.NewRule("list", "watch").Groups(storageGroup).Resources("storageclasses").RuleOrDie(),
+		rbacv1helpers.NewRule("get", "list", "watch", "create", "update", "patch").Groups(legacyGroup).Resources("events").RuleOrDie(),
+		rbacv1helpers.NewRule("get", "list", "watch").Groups(legacyGroup).Resources("nodes").RuleOrDie(),
+	}
+	if utilfeature.DefaultFeatureGate.Enabled(features.KubeletPluginsWatcher) {
+		externalProvisionerRules = append(externalProvisionerRules, rbacv1helpers.NewRule("get", "watch", "list").Groups("csi.storage.k8s.io").Resources("csinodeinfos").RuleOrDie())
+	}
+	roles = append(roles, rbacv1.ClusterRole{
+		// a role for the csi external provisioner
+		ObjectMeta: metav1.ObjectMeta{Name: "system:csi-external-provisioner"},
+		Rules:      externalProvisionerRules,
+	})
+
 	addClusterRoleLabel(roles)
 	return roles
 }

plugin/pkg/auth/authorizer/rbac/bootstrappolicy/testdata/cluster-roles.yaml

@@ -523,6 +523,22 @@ items:
     - patch
     - update
     - watch
+  - apiGroups:
+    - ""
+    resources:
+    - nodes
+    verbs:
+    - get
+    - list
+    - watch
+  - apiGroups:
+    - csi.storage.k8s.io
+    resources:
+    - csinodeinfos
+    verbs:
+    - get
+    - list
+    - watch
 - apiVersion: rbac.authorization.k8s.io/v1
   kind: ClusterRole
   metadata:
@@ -962,6 +978,16 @@ items:
     - volumeattachments
     verbs:
     - get
+  - apiGroups:
+    - csi.storage.k8s.io
+    resources:
+    - csinodeinfos
+    verbs:
+    - create
+    - delete
+    - get
+    - patch
+    - update
 - apiVersion: rbac.authorization.k8s.io/v1
   kind: ClusterRole
   metadata:

staging/src/k8s.io/csi-api/pkg/client/listers/csi/v1alpha1/BUILD

 load("@io_bazel_rules_go//go:def.bzl", "go_library")
 
-filegroup(
-    name = "package-srcs",
-    srcs = glob(["**"]),
-    tags = ["automanaged"],
-    visibility = ["//visibility:private"],
-)
-
-filegroup(
-    name = "all-srcs",
-    srcs = [":package-srcs"],
-    tags = ["automanaged"],
-    visibility = ["//visibility:public"],
-)
-
 go_library(
     name = "go_default_library",
     srcs = [
@@ -31,3 +17,17 @@ go_library(
         "//staging/src/k8s.io/csi-api/pkg/apis/csi/v1alpha1:go_default_library",
     ],
 )
+
+filegroup(
+    name = "package-srcs",
+    srcs = glob(["**"]),
+    tags = ["automanaged"],
+    visibility = ["//visibility:private"],
+)
+
+filegroup(
+    name = "all-srcs",
+    srcs = [":package-srcs"],
+    tags = ["automanaged"],
+    visibility = ["//visibility:public"],
+)

test/integration/auth/node_test.go

@@ -572,6 +572,8 @@ func TestNodeAuthorizer(t *testing.T) {
 	expectForbidden(t, updateNode1Lease(node2Client))
 	expectForbidden(t, patchNode1Lease(node2Client))
 	expectForbidden(t, deleteNode1Lease(node2Client))
+
+	// TODO (verult) CSINodeInfo tests (issue #68254)
 }
 
 // expect executes a function a set number of times until it either returns the