Removing topology feature gate

cli/cmd/images.go

@@ -141,7 +141,6 @@ func getInstallYaml(semVersion *versionutils.Version) (string, error) {
 		UseIPv6:                 false,
 		SilenceAutosupport:      true,
 		Version:                 semVersion,
-		TopologyEnabled:         false,
 		HTTPRequestTimeout:      tridentconfig.HTTPTimeoutString,
 		ServiceAccountName:      getControllerRBACResourceName(),
 	}

cli/cmd/install.go

@@ -581,11 +581,6 @@ func prepareYAMLFiles() error {
 	daemonSetlabels := make(map[string]string)
 	daemonSetlabels[appLabelKey] = TridentNodeLabelValue
 
-	topologyEnabled, err := client.IsTopologyInUse()
-	if err != nil {
-		return fmt.Errorf("could not determine node topology; %v", err)
-	}
-
 	namespaceYAML := k8sclient.GetNamespaceYAML(TridentPodNamespace)
 	if err = writeFile(namespacePath, namespaceYAML); err != nil {
 		return fmt.Errorf("could not write namespace YAML file; %v", err)
@@ -671,7 +666,6 @@ func prepareYAMLFiles() error {
 		UseIPv6:                 useIPv6,
 		SilenceAutosupport:      silenceAutosupport,
 		Version:                 client.ServerVersion(),
-		TopologyEnabled:         topologyEnabled,
 		HTTPRequestTimeout:      httpRequestTimeout.String(),
 		ServiceAccountName:      getControllerRBACResourceName(),
 		ImagePullPolicy:         imagePullPolicy,
@@ -867,8 +861,6 @@ func installTrident() (returnError error) {
 		return fmt.Errorf("not able to connect to Kubernetes API server")
 	}
 
-	topologyEnabled, err := client.IsTopologyInUse()
-
 	// Ensure CSI Trident isn't already installed
 	if installed, namespace, err := isCSITridentInstalled(); err != nil {
 		return fmt.Errorf("could not check if CSI Trident deployment exists; %v", err)
@@ -989,10 +981,6 @@ func installTrident() (returnError error) {
 		return
 	}
 
-	if err != nil {
-		return fmt.Errorf("could not determine node topology; %v", err)
-	}
-
 	// Create the CSI Driver object
 	returnError = createK8SCSIDriver()
 	if returnError != nil {
@@ -1124,7 +1112,6 @@ func installTrident() (returnError error) {
 			UseIPv6:                 useIPv6,
 			SilenceAutosupport:      silenceAutosupport,
 			Version:                 client.ServerVersion(),
-			TopologyEnabled:         topologyEnabled,
 			HTTPRequestTimeout:      httpRequestTimeout.String(),
 			ServiceAccountName:      getControllerRBACResourceName(),
 			ImagePullPolicy:         imagePullPolicy,

cli/k8s_client/k8s_client.go

@@ -2733,22 +2733,6 @@ func (k *KubeClient) deleteOptions() metav1.DeleteOptions {
 	}
 }
 
-func (k *KubeClient) IsTopologyInUse() (bool, error) {
-	nodes, err := k.clientset.CoreV1().Nodes().List(reqCtx(), metav1.ListOptions{})
-	if err != nil {
-		return false, err
-	}
-	for _, node := range nodes.Items {
-		for key := range node.Labels {
-			if strings.Contains(key, "topology.kubernetes.io") {
-				return true, nil
-			}
-		}
-	}
-
-	return false, nil
-}
-
 // addFinalizerToCRDObject is a helper function that updates the CRD object to include our Trident finalizer (
 // definitions are not namespaced)
 func (k *KubeClient) addFinalizerToCRDObject(

cli/k8s_client/types.go

@@ -135,7 +135,6 @@ type KubernetesClient interface {
 	AddFinalizerToCRD(crdName string) error
 	AddFinalizerToCRDs(CRDnames []string) error
 	RemoveFinalizerFromCRD(crdName string) error
-	IsTopologyInUse() (bool, error)
 	GetPersistentVolumes() ([]v1.PersistentVolume, error)
 	GetPersistentVolumeClaims(allNamespaces bool) ([]v1.PersistentVolumeClaim, error)
 	GetVolumeSnapshotClasses() ([]snapshotv1.VolumeSnapshotClass, error)
@@ -164,7 +163,6 @@ type DeploymentYAMLArguments struct {
 	UseIPv6                 bool                  `json:"useIPv6"`
 	SilenceAutosupport      bool                  `json:"silenceAutosupport"`
 	Version                 *versionutils.Version `json:"version"`
-	TopologyEnabled         bool                  `json:"topologyEnabled"`
 	DisableAuditLog         bool                  `json:"disableAuditLog"`
 	HTTPRequestTimeout      string                `json:"httpRequestTimeout"`
 	NodeSelector            map[string]string     `json:"nodeSelector"`

cli/k8s_client/yaml_factory.go

@@ -444,10 +444,6 @@ func GetCSIDeploymentYAML(args *DeploymentYAMLArguments) string {
 	if args.AutosupportHostname != "" {
 		autosupportHostnameLine = fmt.Sprint("- -hostname=", args.AutosupportHostname)
 	}
-	provisionerFeatureGates := ""
-	if args.TopologyEnabled {
-		provisionerFeatureGates = "- --feature-gates=Topology=True"
-	}
 
 	if args.Labels == nil {
 		args.Labels = make(map[string]string)
@@ -517,7 +513,6 @@ func GetCSIDeploymentYAML(args *DeploymentYAMLArguments) string {
 	deploymentYAML = strings.ReplaceAll(deploymentYAML, "{AUTOSUPPORT_DEBUG}", autosupportDebugLine)
 	deploymentYAML = strings.ReplaceAll(deploymentYAML, "{AUTOSUPPORT_SILENCE}",
 		strconv.FormatBool(args.SilenceAutosupport))
-	deploymentYAML = strings.ReplaceAll(deploymentYAML, "{PROVISIONER_FEATURE_GATES}", provisionerFeatureGates)
 	deploymentYAML = strings.ReplaceAll(deploymentYAML, "{HTTP_REQUEST_TIMEOUT}", args.HTTPRequestTimeout)
 	deploymentYAML = strings.ReplaceAll(deploymentYAML, "{SERVICE_ACCOUNT}", args.ServiceAccountName)
 	deploymentYAML = strings.ReplaceAll(deploymentYAML, "{IMAGE_PULL_POLICY}", args.ImagePullPolicy)
@@ -655,7 +650,7 @@ spec:
         - name: asup-dir
           mountPath: /asup
       - name: csi-provisioner
-        image: {CSI_SIDECAR_REGISTRY}/csi-provisioner:v4.0.1
+        image: {CSI_SIDECAR_REGISTRY}/csi-provisioner:v5.1.0
         imagePullPolicy: {IMAGE_PULL_POLICY}
         securityContext:
           capabilities:
@@ -667,7 +662,6 @@ spec:
         - "--csi-address=$(ADDRESS)"
         - "--retry-interval-start=8s"
         - "--retry-interval-max=30s"
-        {PROVISIONER_FEATURE_GATES}
         {K8S_API_CLIENT_SIDECAR_THROTTLE}
         env:
         - name: ADDRESS

cli/k8s_client/yaml_factory_test.go

@@ -91,7 +91,6 @@ func TestYAMLFactory(t *testing.T) {
 		UseIPv6:                 false,
 		SilenceAutosupport:      false,
 		Version:                 version,
-		TopologyEnabled:         false,
 		HTTPRequestTimeout:      config.HTTPTimeoutString,
 		EnableACP:               true,
 		IdentityLabel:           true,
@@ -170,7 +169,6 @@ func TestValidateGetCSIDeploymentYAMLSuccess(t *testing.T) {
 		ControllingCRDetails:    map[string]string{},
 		Version:                 version,
 		HTTPRequestTimeout:      config.HTTPTimeoutString,
-		TopologyEnabled:         true,
 		UseIPv6:                 true,
 		SilenceAutosupport:      false,
 		EnableACP:               true,
@@ -219,7 +217,6 @@ func TestValidateGetCSIDeploymentYAMLFail(t *testing.T) {
 		UseIPv6:                 true,
 		SilenceAutosupport:      false,
 		Version:                 version,
-		TopologyEnabled:         true,
 		HTTPRequestTimeout:      config.HTTPTimeoutString,
 	}
 

frontend/csi/config.go

@@ -11,4 +11,7 @@ const (
 	// CSI supported features
 	CSIBlockVolumes  controllerhelpers.Feature = "CSI_BLOCK_VOLUMES"
 	ExpandCSIVolumes controllerhelpers.Feature = "EXPAND_CSI_VOLUMES"
+
+	// Kubernetes topology labels
+	K8sTopologyRegionLabel = "topology.kubernetes.io/region"
 )

frontend/csi/controller_helpers/kubernetes/plugin.go

@@ -1338,3 +1338,31 @@ func (h *helper) SupportsFeature(ctx context.Context, feature controllerhelpers.
 		return false
 	}
 }
+
+func (h *helper) IsTopologyInUse(ctx context.Context) bool {
+	Logc(ctx).Trace(">>>> IsTopologyInUse")
+	defer Logc(ctx).Trace("<<<< IsTopologyInUse")
+
+	// Get one node with a region topology label.
+	listOpts := metav1.ListOptions{
+		LabelSelector: csi.K8sTopologyRegionLabel,
+		Limit:         1,
+	}
+
+	nodes, err := h.kubeClient.CoreV1().Nodes().List(ctx, listOpts)
+	if err != nil {
+		Logc(ctx).WithError(err).Error("Failed to list nodes with topology label. Assuming topology in use to be 'false' by default.")
+		return false
+	}
+
+	// If there exists even a single node with topology label, we consider topology to be in use.
+	topologyInUse := false
+	if nodes != nil && len(nodes.Items) > 0 {
+		topologyInUse = true
+	}
+
+	fields := LogFields{"topologyInUse": topologyInUse}
+	Logc(ctx).WithFields(fields).Info("Successfully determined if topology is in use.")
+
+	return topologyInUse
+}

frontend/csi/controller_helpers/kubernetes/plugin_test.go

@@ -1409,3 +1409,69 @@ func TestValidateStorageClassParameters(t *testing.T) {
 		})
 	}
 }
+
+func TestIsTopologyInUse(t *testing.T) {
+	ctx := context.TODO()
+	_, plugin := newMockPlugin(t)
+
+	tt := map[string]struct {
+		labels      map[string]string
+		injectError bool
+		expected    bool
+	}{
+		"node with nil labels": {
+			labels:   nil,
+			expected: false,
+		},
+		"node with empty labels": {
+			labels:   map[string]string{},
+			expected: false,
+		},
+		"node with labels, but no topology labels": {
+			labels:   map[string]string{"hostname.kubernetes.io/name": "host1"},
+			expected: false,
+		},
+		"node with non-region topology label": {
+			labels:   map[string]string{"topology.kubernetes.io/zone": "zone1"},
+			expected: false,
+		},
+		"node with multiple topology labels": {
+			labels:   map[string]string{"topology.kubernetes.io/region": "region1", "topology.kubernetes.io/zone": "zone1"},
+			expected: true,
+		},
+		"error while listing the nodes": {
+			labels:      map[string]string{"topology.kubernetes.io/region": "region1", "topology.kubernetes.io/zone": "zone1"},
+			injectError: true,
+			expected:    false,
+		},
+	}
+
+	for name, test := range tt {
+		t.Run(name, func(t *testing.T) {
+			// create fake nodes and add to a fake k8s client
+			fakeNode := &v1.Node{ObjectMeta: metav1.ObjectMeta{Name: "fakeNode", Labels: test.labels}}
+			clientSet := k8sfake.NewSimpleClientset(fakeNode)
+
+			// add reactor to either return the list or return error if required
+			clientSet.Fake.PrependReactor(
+				"list" /* use '*' for all operations */, "*", /* use '*' all object types */
+				func(_ k8stesting.Action) (handled bool, ret runtime.Object, err error) {
+					if test.injectError {
+						status := &k8serrors.StatusError{ErrStatus: metav1.Status{Reason: metav1.StatusFailure}}
+						return true, nil, status
+					} else {
+						return true, &v1.NodeList{Items: []v1.Node{*fakeNode}}, nil
+					}
+				},
+			)
+
+			// add the fake client to the plugin
+			plugin.kubeClient = clientSet
+
+			// check if the topology is in use
+			result := plugin.IsTopologyInUse(ctx)
+
+			assert.Equal(t, test.expected, result, fmt.Sprintf("topology usage not as expected; expected %v, got %v", test.expected, result))
+		})
+	}
+}

frontend/csi/controller_helpers/plain/plugin.go

@@ -166,3 +166,7 @@ func (h *helper) SupportsFeature(_ context.Context, feature controllerhelpers.Fe
 		return false
 	}
 }
+
+func (h *helper) IsTopologyInUse(_ context.Context) bool {
+	return false
+}

frontend/csi/controller_helpers/plain/plugin_test.go

@@ -309,3 +309,17 @@ func TestSupportsFeature(t *testing.T) {
 		assert.Equal(t, tc.expected, supported, "Feature is not supported")
 	}
 }
+
+func TestIsTopologyInUse(t *testing.T) {
+	mockCtrl := gomock.NewController(t)
+	orchestrator := mock.NewMockOrchestrator(mockCtrl)
+	p := NewHelper(orchestrator)
+	plugin, ok := p.(controller_helpers.ControllerHelper)
+	if !ok {
+		t.Fatal("Could not cast the helper to a ControllerHelper!")
+	}
+
+	result := plugin.IsTopologyInUse(context.TODO())
+
+	assert.Equal(t, false, result, "expected topology usage to be false")
+}

frontend/csi/controller_helpers/types.go

@@ -71,4 +71,7 @@ type ControllerHelper interface {
 	// Version returns the version of the CO this helper is managing, or the supported
 	// CSI version in the plain-CSI case.  This value is reported in Trident's telemetry.
 	Version() string
+
+	// IsTopologyInUse checks if any node in the cluster has topology labels
+	IsTopologyInUse(ctx context.Context) bool
 }

frontend/csi/identity_server.go

@@ -55,26 +55,34 @@ func (p *Plugin) GetPluginCapabilities(
 	ctx = SetContextWorkflow(ctx, WorkflowIdentityGetCapabilities)
 	ctx = GenerateRequestContextForLayer(ctx, LogLayerCSIFrontend)
 
-	fields := LogFields{"Method": "GetPluginCapabilities", "Type": "CSI_Identity"}
+	fields := LogFields{"Method": "GetPluginCapabilities", "Type": "CSI_Identity", "topologyInUse": p.topologyInUse}
 	Logc(ctx).WithFields(fields).Trace(">>>> GetPluginCapabilities")
 	defer Logc(ctx).WithFields(fields).Trace("<<<< GetPluginCapabilities")
 
-	return &csi.GetPluginCapabilitiesResponse{
-		Capabilities: []*csi.PluginCapability{
-			{
-				Type: &csi.PluginCapability_Service_{
-					Service: &csi.PluginCapability_Service{
-						Type: csi.PluginCapability_Service_CONTROLLER_SERVICE,
-					},
+	// Add controller service capability
+	csiPluginCap := []*csi.PluginCapability{
+		{
+			Type: &csi.PluginCapability_Service_{
+				Service: &csi.PluginCapability_Service{
+					Type: csi.PluginCapability_Service_CONTROLLER_SERVICE,
 				},
 			},
-			{
-				Type: &csi.PluginCapability_Service_{
-					Service: &csi.PluginCapability_Service{
-						Type: csi.PluginCapability_Service_VOLUME_ACCESSIBILITY_CONSTRAINTS,
-					},
+		},
+	}
+
+	// If topology is in use, add VOLUME_ACCESSIBILITY_CONSTRAINTS capability
+	if p.topologyInUse {
+		Logc(ctx).WithFields(fields).Info("Topology is in use. Adding VOLUME_ACCESSIBILITY_CONSTRAINTS capability.")
+		csiPluginCap = append(csiPluginCap, &csi.PluginCapability{
+			Type: &csi.PluginCapability_Service_{
+				Service: &csi.PluginCapability_Service{
+					Type: csi.PluginCapability_Service_VOLUME_ACCESSIBILITY_CONSTRAINTS,
 				},
 			},
-		},
+		})
+	}
+
+	return &csi.GetPluginCapabilitiesResponse{
+		Capabilities: csiPluginCap,
 	}, nil
 }

frontend/csi/node_server.go

@@ -761,6 +761,19 @@ func (p *Plugin) nodeRegisterWithController(ctx context.Context, timeout time.Du
 			topologyLabels = nodeDetails.TopologyLabels
 			node.TopologyLabels = nodeDetails.TopologyLabels
 
+			// Assume topology to be in use only if there exists a topology label with key "topology.kubernetes.io/region" on the node.
+			if len(topologyLabels) > 0 {
+				val, ok := topologyLabels[K8sTopologyRegionLabel]
+				fields := LogFields{"topologyLabelKey": K8sTopologyRegionLabel, "value": val}
+				if ok && val != "" {
+					Logc(ctx).WithFields(fields).Infof("%s label found on node. Assuming topology to be in use.", K8sTopologyRegionLabel)
+					p.topologyInUse = true
+				} else {
+					Logc(ctx).WithFields(fields).Infof("%s label not found on node. Assuming topology not in use.", K8sTopologyRegionLabel)
+					p.topologyInUse = false
+				}
+			}
+
 			// Setting log level, log workflows and log layers on the node same as to what is set on the controller.
 			if err = p.orchestrator.SetLogLevel(ctx, nodeDetails.LogLevel); err != nil {
 				Logc(ctx).WithError(err).Error("Unable to set log level.")