Add flag to set volume attachment limit for trident csiNodes

This introduces a new flag `-volume_attach_limit`, meaningful when running
trident_orchestrator for a node server. Allows to specify a maximum capacity of
allocatable volumes per csi node. Default value (0) means that no cap is set.

Example output from `kubectl describe csinodes.storage.k8s.io` when running
trident daemonset with `-volume_attach_limit=10`:

```
Spec:
  Drivers:
    csi.trident.netapp.io:
      Node ID:  worker-0.exp-1.merit.uw.systems
      Allocatables:
        Count:        10
      Topology Keys:  [topology.kubernetes.io/zone]
```

Trying to scale more than 10 volumes (following the above example) in the same
node will not be allowed. Pods that will not fit will remain `Pending` and
checking events will yield something like:

```
Events:
  Type     Reason            Age   From               Message
  ----     ------            ----  ----               -------
  Warning  FailedScheduling  32s   default-scheduler  0/8 nodes are available: 1 node(s) exceed max volume count, 2 node(s) had taint {node-role.kubernetes.io/master: }, that the pod didn't tolerate, 5 node(s) didn't match Pod's node affinity/selector.
```

frontend/csi/node_server.go

@@ -416,7 +416,7 @@ func (p *Plugin) NodeGetInfo(
 		Segments: topologyLabels,
 	}
 
-	return &csi.NodeGetInfoResponse{NodeId: p.nodeName, AccessibleTopology: topology}, nil
+	return &csi.NodeGetInfoResponse{NodeId: p.nodeName, MaxVolumesPerNode: p.volumeAttachLimit, AccessibleTopology: topology}, nil
 }
 
 func (p *Plugin) nodeGetInfo(ctx context.Context) *utils.Node {

frontend/csi/plugin.go

@@ -36,6 +36,8 @@ type Plugin struct {
 	endpoint string
 	role     string
 
+	volumeAttachLimit int64
+
 	unsafeDetach bool
 
 	hostInfo *utils.HostSystem
@@ -105,22 +107,23 @@ func NewControllerPlugin(
 }
 
 func NewNodePlugin(
-	nodeName, endpoint, caCert, clientCert, clientKey, aesKeyFile string, orchestrator core.Orchestrator,
+	nodeName, endpoint, caCert, clientCert, clientKey, aesKeyFile string, volumeAttachLimit int64, orchestrator core.Orchestrator,
 	unsafeDetach, nodePrep bool,
 ) (*Plugin, error) {
 
 	ctx := GenerateRequestContext(context.Background(), "", ContextSourceInternal)
 
 	p := &Plugin{
-		orchestrator: orchestrator,
-		name:         Provisioner,
-		nodeName:     nodeName,
-		version:      tridentconfig.OrchestratorVersion.ShortString(),
-		endpoint:     endpoint,
-		role:         CSINode,
-		unsafeDetach: unsafeDetach,
-		opCache:      sync.Map{},
-		nodePrep:     &utils.NodePrep{Enabled: nodePrep},
+		orchestrator:      orchestrator,
+		name:              Provisioner,
+		nodeName:          nodeName,
+		version:           tridentconfig.OrchestratorVersion.ShortString(),
+		endpoint:          endpoint,
+		role:              CSINode,
+		unsafeDetach:      unsafeDetach,
+		opCache:           sync.Map{},
+		nodePrep:          &utils.NodePrep{Enabled: nodePrep},
+		volumeAttachLimit: volumeAttachLimit,
 	}
 
 	// Initialize node prep statuses

main.go

@@ -90,6 +90,8 @@ var (
 
 	aesKey = flag.String("aes_key", config.AESKeyPath, "AES encryption key")
 
+	volumeAttachLimit = flag.Int64("volume_attach_limit", 0, "Maximum number of volumes that controller can publish to the node, 0 (default) means unlimited")
+
 	// HTTP metrics interface
 	metricsAddress = flag.String("metrics_address", "", "Storage orchestrator metrics address")
 	metricsPort    = flag.String("metrics_port", "8001", "Storage orchestrator metrics port")
@@ -347,7 +349,7 @@ func main() {
 			csiFrontend, err = csi.NewControllerPlugin(*csiNodeName, *csiEndpoint, *aesKey, orchestrator, &hybridPlugin)
 		case csi.CSINode:
 			csiFrontend, err = csi.NewNodePlugin(*csiNodeName, *csiEndpoint, *httpsCACert, *httpsClientCert,
-				*httpsClientKey, *aesKey, orchestrator, *csiUnsafeNodeDetach, *nodePrep)
+				*httpsClientKey, *aesKey, *volumeAttachLimit, orchestrator, *csiUnsafeNodeDetach, *nodePrep)
 			enableMutualTLS = false
 			handler = rest.NewNodeRouter(csiFrontend)
 		case csi.CSIAllInOne: