Repository: kubernetes-sigs/gcp-filestore-csi-driver
Pull request #461 (2023-03-24 08:59:25)

Implement configurable max shares CreateVolume path with feature gate
<!--  Thanks for sending a pull request!  Here are some tips for you:

1. If this is your first time, please read our contributor guidelines: https://git.k8s.io/community/contributors/guide#your-first-contribution and developer guide https://git.k8s.io/community/contributors/devel/development.md#development-guide
2. Follow the instructions for writing a release note: https://git.k8s.io/community/contributors/guide/release-notes.md
3. If the PR is unfinished, see how to mark it: https://git.k8s.io/community/contributors/guide/pull-requests.md#marking-unfinished-pull-requests
-->

**What type of PR is this?**
> Uncomment only one ` /kind <>` line, hit enter to put that in a new line, and remove leading whitespaces from that line:
>
> /kind api-change
> /kind bug
> /kind cleanup
> /kind design
> /kind documentation
> /kind failing-test
/kind feature
> /kind flake

**What this PR does / why we need it**:
Second PR in a series of PR to implement the configurable max shares/instance feature. In this PR, the CreateVolume code path is implemented and gated under the feature flag.
Key changes:
1. validate parse the storage class config, and generate the max supported shares. 
2. Plumb the max shares to the instance create call
3. Implement the description override flag to setup maxshares, min share size knobs (which is needed until filestore is ready with instance.MaxSharesCount field)
4. If feature is disabled, then maxShareCount is set to 0. This translates to a default value of 10 shares/instance in filestore.

**Which issue(s) this PR fixes**:
<!--
*Automatically closes linked issue when PR is merged.
Usage: `Fixes #<issue number>`, or `Fixes (paste link of issue)`.
_If PR is about `failing-tests or flakes`, please post the related issues/tests in a comment and do not use `Fixes`_*
-->
Fixes #

**Special notes for your reviewer**:
Future PRs will include
1. ControllerExpandVolume [changes](https://docs.google.com/document/d/1Jsx2H90NPjP8kddrljwwi0LKIlAmkT_ImAa_f4mwUfI/edit#bookmark=id.o8s7hdo9or5a)
2. Poll Timeout [changes](https://docs.google.com/document/d/1Jsx2H90NPjP8kddrljwwi0LKIlAmkT_ImAa_f4mwUfI/edit#bookmark=id.f59tfd9c6m8)


For this PR, 
1. tested manually that feature disabled path works as expected. Instance maxShareCount returns 10.
2. Tested description override. Instance maxShareCount returns value according to the description override knob.

Yet to test instance.maxShareCount during create. Needs an allowlist for that.

**Does this PR introduce a user-facing change?**:
<!--
If no, just write "NONE" in the release-note block below.
If yes, a release note is required:
Enter your extended release note in the block below. If the PR requires additional action from users switching to the new release, include the string "action required".
-->
```release-note
Implement configurable max shares CreateVolume path with feature gate
```


k8s-ci-robot: [APPROVALNOTIFIER] This PR is **APPROVED**

This pull-request has been approved by: *<a href="https://github.com/kubernetes-sigs/gcp-filestore-csi-driver/pull/461#" title="Author self-approved">saikat-royc</a>*

The full list of commands accepted by this bot can be found [here](https://go.k8s.io/bot-commands?repo=kubernetes-sigs%2Fgcp-filestore-csi-driver).

The pull request process is described [here](https://git.k8s.io/community/contributors/guide/owners.md#the-code-review-process)

<details >
Needs approval from an approver in each of these files:

- ~~[OWNERS](https://github.com/kubernetes-sigs/gcp-filestore-csi-driver/blob/master/OWNERS)~~ [saikat-royc]

Approvers can indicate their approval by writing `/approve` in a comment
Approvers can cancel approval by writing `/approve cancel` in a comment
</details>
<!-- META={"approvers":[]} -->
saikat-royc: /hold
saikat-royc: /assign @leiyiz 
k8s-ci-robot: @saikat-royc: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-gcp-filestore-csi-driver-kubernetes-integration | 97905e636313e37607f62e67f548c2efa0136289 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/461/pull-gcp-filestore-csi-driver-kubernetes-integration/1639224023097282560/) | true | `/test pull-gcp-filestore-csi-driver-kubernetes-integration`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=461). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Asaikat-royc). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/gcp-filestore-csi-driver/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->
saikat-royc: /retest
k8s-ci-robot: @saikat-royc: The following test **failed**, say `/retest` to rerun all failed tests or `/retest-required` to rerun all mandatory failed tests:

Test name | Commit | Details | Required | Rerun command
--- | --- | --- | --- | ---
pull-gcp-filestore-csi-driver-verify | 5719d6d43d5909b4d99c730e9b1af3e5bcfc8bf5 | [link](https://prow.k8s.io/view/gs/kubernetes-jenkins/pr-logs/pull/sigs.k8s.io_gcp-filestore-csi-driver/461/pull-gcp-filestore-csi-driver-verify/1641895657771896832/) | true | `/test pull-gcp-filestore-csi-driver-verify`

[Full PR test history](https://prow.k8s.io/pr-history?org=kubernetes-sigs&repo=gcp-filestore-csi-driver&pr=461). [Your PR dashboard](https://prow.k8s.io/pr?query=is%3Apr%20state%3Aopen%20author%3Asaikat-royc). Please help us cut down on flakes by [linking to](https://git.k8s.io/community/contributors/devel/sig-testing/flaky-tests.md#filing-issues-for-flaky-tests) an [open issue](https://github.com/kubernetes-sigs/gcp-filestore-csi-driver/issues?q=is:issue+is:open) when you hit one in your PR.

<details>

Instructions for interacting with me using PR comments are available [here](https://git.k8s.io/community/contributors/guide/pull-requests.md).  If you have questions or suggestions related to my behavior, please file an issue against the [kubernetes/test-infra](https://github.com/kubernetes/test-infra/issues/new?title=Prow%20issue:) repository. I understand the commands that are listed [here](https://go.k8s.io/bot-commands).
</details>
<!-- test report -->

Patch:
diff --git a/pkg/cloud_provider/file/file.go b/pkg/cloud_provider/file/file.go
index 8712e175..0b04bbe6 100644
--- a/pkg/cloud_provider/file/file.go
+++ b/pkg/cloud_provider/file/file.go
@@ -71,6 +71,7 @@ type MultishareInstance struct {
 	State              string
 	KmsKeyName         string
 	Description        string
+	MaxShareCount      int
 }
 
 func (i *MultishareInstance) String() string {
@@ -803,10 +804,11 @@ func (manager *gcfsServiceManager) StartCreateMultishareInstanceOp(ctx context.C
 				ConnectMode:     instance.Network.ConnectMode,
 			},
 		},
-		CapacityGb:  util.BytesToGb(instance.CapacityBytes),
-		KmsKeyName:  instance.KmsKeyName,
-		Labels:      instance.Labels,
-		Description: instance.Description,
+		CapacityGb:    util.BytesToGb(instance.CapacityBytes),
+		KmsKeyName:    instance.KmsKeyName,
+		Labels:        instance.Labels,
+		Description:   instance.Description,
+		MaxShareCount: int64(instance.MaxShareCount),
 	}
 
 	op, err := manager.multishareInstancesService.Create(locationURI(instance.Project, instance.Location), targetinstance).InstanceId(instance.Name).Context(ctx).Do()
@@ -1088,6 +1090,7 @@ func cloudInstanceToMultishareInstance(instance *filev1beta1multishare.Instance)
 		MaxCapacityBytes:   instance.MaxCapacityGb * util.Gb,
 		CapacityStepSizeGb: instance.CapacityStepSizeGb,
 		Description:        instance.Description,
+		MaxShareCount:      int(instance.MaxShareCount),
 	}, nil
 }
 
diff --git a/pkg/csi_driver/multishare_controller.go b/pkg/csi_driver/multishare_controller.go
index f53e287f..48489904 100644
--- a/pkg/csi_driver/multishare_controller.go
+++ b/pkg/csi_driver/multishare_controller.go
@@ -18,12 +18,14 @@ package driver
 
 import (
 	"fmt"
+	"strconv"
 	"strings"
 
 	csi "github.com/container-storage-interface/spec/lib/go/csi"
 	"golang.org/x/net/context"
 	"google.golang.org/grpc/codes"
 	"google.golang.org/grpc/status"
+	"k8s.io/apimachinery/pkg/api/resource"
 	"k8s.io/apimachinery/pkg/util/uuid"
 	"k8s.io/klog/v2"
 	cloud "sigs.k8s.io/gcp-filestore-csi-driver/pkg/cloud_provider"
@@ -39,6 +41,9 @@ const (
 	methodExpandVolume              = "ExpandVolume"
 	ecfsDataPlaneVersionFormat      = "GoogleReserved-CustomVMImage=clh.image.ems.path:projects/%s/global/images/ems-filestore-scaleout-%s"
 	ecfsCustom100sharesConfigFormat = "GoogleReservedOverrides={\"CustomMultiShareConfig\":{\"MaxShareCount\": %d, \"MinShareSizeGB\":%d}}"
+
+	// volume context attributes
+	attrMaxShareSize = "max-share-size"
 )
 
 // MultishareController handles CSI calls for volumes which use Filestore multishare instances.
@@ -94,7 +99,17 @@ func (m *MultishareController) CreateVolume(ctx context.Context, req *csi.Create
 		return nil, status.Error(codes.InvalidArgument, err.Error())
 	}
 
-	reqBytes, err := getShareRequestCapacity(req.GetCapacityRange())
+	maxSharesPerInstance, maxShareSizeSizeBytes, err := m.parseMaxVolumeSizeParam(req)
+	if err != nil {
+		return nil, status.Error(codes.InvalidArgument, err.Error())
+	}
+
+	var reqBytes int64
+	if m.featureMaxSharePerInstance {
+		reqBytes, err = getShareRequestCapacity(req.GetCapacityRange(), util.MinShareSizeConfigurableBytes, maxShareSizeSizeBytes)
+	} else {
+		reqBytes, err = getShareRequestCapacity(req.GetCapacityRange(), util.MinShareSizeBytes, util.MaxShareSizeBytes)
+	}
 	if err != nil {
 		return nil, status.Error(codes.InvalidArgument, err.Error())
 	}
@@ -107,18 +122,30 @@ func (m *MultishareController) CreateVolume(ctx context.Context, req *csi.Create
 	defer m.volumeLocks.Release(name)
 
 	// If no eligible instance found, the ops manager may decide to create a new instance. Prepare a multishare instacne object for such a scenario.
-	instance, err := m.generateNewMultishareInstance(util.NewMultishareInstancePrefix+string(uuid.NewUUID()), req)
+	instance, err := m.generateNewMultishareInstance(util.NewMultishareInstancePrefix+string(uuid.NewUUID()), req, maxSharesPerInstance)
 	if err != nil {
 		return nil, err
 	}
 
+	if m.featureMaxSharePerInstance && m.descOverrideMaxSharesPerInstance != "" && m.descOverrideMinShareSizeBytes != "" {
+		sharesPerInstance, err := strconv.Atoi(m.descOverrideMaxSharesPerInstance)
+		if err != nil {
+			return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("invalid description override value %s", m.descOverrideMaxSharesPerInstance))
+		}
+		minShareSizeGB, err := strconv.Atoi(m.descOverrideMinShareSizeBytes)
+		if err != nil {
+			return nil, status.Error(codes.InvalidArgument, fmt.Sprintf("invalid description override value %s", m.descOverrideMinShareSizeBytes))
+		}
+		instance.Description = fmt.Sprintf(ecfsCustom100sharesConfigFormat, sharesPerInstance, minShareSizeGB)
+	}
+
 	workflow, share, err := m.opsManager.setupEligibleInstanceAndStartWorkflow(ctx, req, instance)
 	if err != nil {
 		return nil, err
 	}
 
 	if share != nil {
-		return m.getShareAndGenerateCSICreateVolumeResponse(ctx, instanceScPrefix, share)
+		return m.getShareAndGenerateCSICreateVolumeResponse(ctx, instanceScPrefix, share, maxShareSizeSizeBytes)
 	}
 
 	// lock released. poll for op.
@@ -130,7 +157,7 @@ func (m *MultishareController) CreateVolume(ctx context.Context, req *csi.Create
 
 	klog.Infof("Poll for operation %s (type %s) completed", workflow.opName, workflow.opType.String())
 	if workflow.opType == util.ShareCreate {
-		return m.getShareAndGenerateCSICreateVolumeResponse(ctx, instanceScPrefix, workflow.share)
+		return m.getShareAndGenerateCSICreateVolumeResponse(ctx, instanceScPrefix, workflow.share, maxShareSizeSizeBytes)
 	}
 
 	var shareCreateWorkflow *Workflow
@@ -155,10 +182,10 @@ func (m *MultishareController) CreateVolume(ctx context.Context, req *csi.Create
 		errCode := file.PollOpErrorCode(err)
 		return nil, status.Errorf(*errCode, "%s operation %q poll error: %v", shareCreateWorkflow.opType.String(), shareCreateWorkflow.opName, err)
 	}
-	return m.getShareAndGenerateCSICreateVolumeResponse(ctx, instanceScPrefix, newShare)
+	return m.getShareAndGenerateCSICreateVolumeResponse(ctx, instanceScPrefix, newShare, maxShareSizeSizeBytes)
 }
 
-func (m *MultishareController) getShareAndGenerateCSICreateVolumeResponse(ctx context.Context, instancePrefix string, s *file.Share) (*csi.CreateVolumeResponse, error) {
+func (m *MultishareController) getShareAndGenerateCSICreateVolumeResponse(ctx context.Context, instancePrefix string, s *file.Share, maxShareSizeSizeBytes int64) (*csi.CreateVolumeResponse, error) {
 	share, err := m.cloud.File.GetShare(ctx, s)
 	if err != nil {
 		return nil, status.Errorf(codes.Internal, err.Error())
@@ -167,7 +194,7 @@ func (m *MultishareController) getShareAndGenerateCSICreateVolumeResponse(ctx co
 	if share.State != "READY" {
 		return nil, status.Errorf(codes.Aborted, "share %s not ready, state %s", share.Name, share.State)
 	}
-	return m.generateCSICreateVolumeResponse(instancePrefix, share)
+	return m.generateCSICreateVolumeResponse(instancePrefix, share, maxShareSizeSizeBytes)
 }
 
 func (m *MultishareController) DeleteVolume(ctx context.Context, req *csi.DeleteVolumeRequest) (*csi.DeleteVolumeResponse, error) {
@@ -263,7 +290,8 @@ func (m *MultishareController) ControllerExpandVolume(ctx context.Context, req *
 		return nil, status.Error(codes.InvalidArgument, "ControllerExpandVolume volume ID must be provided")
 	}
 
-	reqBytes, err := getShareRequestCapacity(req.GetCapacityRange())
+	// TODO: This logic will change to derive the max allowable share size from the PV.spec.CSI.volumeAttributes[“max-share-size”]
+	reqBytes, err := getShareRequestCapacity(req.GetCapacityRange(), util.MinShareSizeConfigurableBytes, util.MaxShareSizeBytes)
 	if err != nil {
 		return nil, status.Error(codes.InvalidArgument, err.Error())
 	}
@@ -374,7 +402,7 @@ func getInstanceSCPrefix(req *csi.CreateVolumeRequest) (string, error) {
 	return v, nil
 }
 
-func (m *MultishareController) generateNewMultishareInstance(instanceName string, req *csi.CreateVolumeRequest) (*file.MultishareInstance, error) {
+func (m *MultishareController) generateNewMultishareInstance(instanceName string, req *csi.CreateVolumeRequest, maxShareCount int) (*file.MultishareInstance, error) {
 	region, err := m.pickRegion(req.GetAccessibilityRequirements())
 	if err != nil {
 		return nil, status.Error(codes.InvalidArgument, err.Error())
@@ -404,6 +432,8 @@ func (m *MultishareController) generateNewMultishareInstance(instanceName string
 			continue
 		case paramMultishareInstanceScLabel:
 			continue
+		case paramMaxVolumeSize:
+			continue
 		case ParameterKeyLabels, ParameterKeyPVCName, ParameterKeyPVCNamespace, ParameterKeyPVName, paramMultishare:
 		case "csiprovisionersecretname", "csiprovisionersecretnamespace":
 		default:
@@ -427,7 +457,7 @@ func (m *MultishareController) generateNewMultishareInstance(instanceName string
 		return nil, status.Errorf(codes.InvalidArgument, err.Error())
 	}
 
-	return &file.MultishareInstance{
+	f := &file.MultishareInstance{
 		Project:       m.cloud.Project,
 		Name:          instanceName,
 		CapacityBytes: util.MinMultishareInstanceSizeBytes,
@@ -440,14 +470,19 @@ func (m *MultishareController) generateNewMultishareInstance(instanceName string
 		KmsKeyName:  kmsKeyName,
 		Labels:      labels,
 		Description: generateInstanceDescFromEcfsDesc(m.ecfsDescription),
-	}, nil
+	}
+	if m.featureMaxSharePerInstance {
+		f.MaxShareCount = maxShareCount
+	}
+	return f, nil
 }
 
 func generateNewShare(name string, parent *file.MultishareInstance, req *csi.CreateVolumeRequest) (*file.Share, error) {
 	if parent == nil {
 		return nil, status.Error(codes.Internal, "parent mulishare instance is empty")
 	}
-	targetSizeBytes, err := getShareRequestCapacity(req.CapacityRange)
+	// The share size request is already validated in CreateVolume call
+	targetSizeBytes, err := getShareRequestCapacity(req.CapacityRange, util.MinShareSizeConfigurableBytes, util.MaxShareSizeBytes)
 	if err != nil {
 		return nil, status.Error(codes.InvalidArgument, err.Error())
 	}
@@ -526,9 +561,9 @@ func extractShareLabels(parameters map[string]string) map[string]string {
 	return shareLabels
 }
 
-func getShareRequestCapacity(capRange *csi.CapacityRange) (int64, error) {
+func getShareRequestCapacity(capRange *csi.CapacityRange, minShareSizeBytes, maxShareSizeBytes int64) (int64, error) {
 	if capRange == nil {
-		return util.MinShareSizeBytes, nil
+		return minShareSizeBytes, nil
 	}
 
 	rCap := capRange.GetRequiredBytes()
@@ -546,22 +581,22 @@ func getShareRequestCapacity(capRange *csi.CapacityRange) (int64, error) {
 
 	// Check bounds of limit and request.
 	if lSet {
-		if lCap < util.MinShareSizeBytes {
-			return 0, status.Errorf(codes.InvalidArgument, "Limit bytes %v is less than minimum share size bytes %v", lCap, util.MinShareSizeBytes)
+		if lCap < minShareSizeBytes {
+			return 0, status.Errorf(codes.InvalidArgument, "Limit bytes %v is less than minimum share size bytes %v", lCap, minShareSizeBytes)
 		}
 
-		if lCap > util.MaxShareSizeBytes {
-			return 0, status.Errorf(codes.InvalidArgument, "Limit bytes %v is greater than maximum share size bytes %v", lCap, util.MaxShareSizeBytes)
+		if lCap > maxShareSizeBytes {
+			return 0, status.Errorf(codes.InvalidArgument, "Limit bytes %v is greater than maximum share size bytes %v", lCap, maxShareSizeBytes)
 		}
 	}
 
 	if rSet {
-		if rCap < util.MinShareSizeBytes {
-			return 0, status.Errorf(codes.InvalidArgument, "Request bytes %v is less than minimum share size bytes %v", rCap, util.MinShareSizeBytes)
+		if rCap < minShareSizeBytes {
+			return 0, status.Errorf(codes.InvalidArgument, "Request bytes %v is less than minimum share size bytes %v", rCap, minShareSizeBytes)
 		}
 
-		if rCap > util.MaxShareSizeBytes {
-			return 0, status.Errorf(codes.InvalidArgument, "Request bytes %v is greater than maximum share size bytes %v", rCap, util.MaxShareSizeBytes)
+		if rCap > maxShareSizeBytes {
+			return 0, status.Errorf(codes.InvalidArgument, "Request bytes %v is greater than maximum share size bytes %v", rCap, maxShareSizeBytes)
 		}
 	}
 
@@ -572,7 +607,7 @@ func getShareRequestCapacity(capRange *csi.CapacityRange) (int64, error) {
 	return rCap, nil
 }
 
-func (m *MultishareController) generateCSICreateVolumeResponse(instancePrefix string, s *file.Share) (*csi.CreateVolumeResponse, error) {
+func (m *MultishareController) generateCSICreateVolumeResponse(instancePrefix string, s *file.Share, maxShareSizeSizeBytes int64) (*csi.CreateVolumeResponse, error) {
 	volId, err := generateMultishareVolumeIdFromShare(instancePrefix, s)
 	if err != nil {
 		return nil, status.Errorf(codes.Internal, err.Error())
@@ -590,6 +625,9 @@ func (m *MultishareController) generateCSICreateVolumeResponse(instancePrefix st
 	if m.driver.config.FeatureOptions.FeatureLockRelease.Enabled {
 		resp.Volume.VolumeContext[attrSupportLockRelease] = "true"
 	}
+	if m.featureMaxSharePerInstance {
+		resp.Volume.VolumeContext[attrMaxShareSize] = strconv.Itoa(int(maxShareSizeSizeBytes))
+	}
 	klog.Infof("CreateVolume resp: %+v", resp)
 	return resp, nil
 }
@@ -639,3 +677,49 @@ func generateInstanceDescFromEcfsDesc(desc string) string {
 	klog.V(4).Infof("generated description for multishare instance %s", d)
 	return d
 }
+
+func (m *MultishareController) parseMaxVolumeSizeParam(req *csi.CreateVolumeRequest) (int, int64, error) {
+	params := req.GetParameters()
+	v, ok := params[paramMaxVolumeSize]
+	if !m.featureMaxSharePerInstance && ok {
+		return 0, 0, fmt.Errorf("configurable max shares per instance feature not enabled")
+	}
+	if !ok {
+		return util.MaxSharesPerInstance, util.MaxShareSizeBytes, nil
+	}
+	if v == "" {
+		return 0, 0, fmt.Errorf("value is empty for %q key", paramMaxVolumeSize)
+	}
+	val, err := resource.ParseQuantity(v)
+	if err != nil {
+		return 0, 0, err
+	}
+
+	valBytes := val.Value()
+	sharesPerInstance, err := getSharesPerInstance(valBytes)
+	if err != nil {
+		return 0, 0, err
+	}
+	return sharesPerInstance, valBytes, nil
+}
+
+func getSharesPerInstance(volSizeBytes int64) (int, error) {
+	if !isValidMaxVolSize(volSizeBytes) {
+		return 0, fmt.Errorf("unsupported max volume size %d, supported sizes: '128Gi', '256Gi', '512Gi', '1024Gi'", volSizeBytes)
+	}
+	return int(util.MaxMultishareInstanceSizeBytes / volSizeBytes), nil
+}
+
+func isValidMaxVolSize(val int64) bool {
+	switch val {
+	case 128 * util.Gb:
+		return true
+	case 256 * util.Gb:
+		return true
+	case 512 * util.Gb:
+		return true
+	case util.Tb:
+		return true
+	}
+	return false
+}
diff --git a/pkg/csi_driver/multishare_ops_manager.go b/pkg/csi_driver/multishare_ops_manager.go
index 4e7c71fe..1344f87b 100644
--- a/pkg/csi_driver/multishare_ops_manager.go
+++ b/pkg/csi_driver/multishare_ops_manager.go
@@ -356,7 +356,7 @@ func (m *MultishareOpsManager) runEligibleInstanceCheck(ctx context.Context, req
 	var nonReadyEligibleInstances []*file.MultishareInstance
 
 	for _, instance := range instances {
-		klog.Infof("Found multishare instance %s/%s/%s with state %s.", instance.Project, instance.Location, instance.Name, instance.State)
+		klog.Infof("Found multishare instance %s/%s/%s with state %s and max share count %d", instance.Project, instance.Location, instance.Name, instance.State, instance.MaxShareCount)
 		if instance.State == "CREATING" || instance.State == "REPAIRING" {
 			klog.Infof("Instance %s/%s/%s with state %s is not ready", instance.Project, instance.Location, instance.Name, instance.State)
 			nonReadyEligibleInstances = append(nonReadyEligibleInstances, instance)
@@ -380,7 +380,14 @@ func (m *MultishareOpsManager) runEligibleInstanceCheck(ctx context.Context, req
 				klog.Errorf("Failed to list shares of instance %s/%s/%s, err:%v", instance.Project, instance.Location, instance.Name, err.Error())
 				return nil, err
 			}
-			if len(shares) >= util.MaxSharesPerInstance {
+
+			// If we encounter a scenario where the configurable shares per Filestore instance feature is disabled, CSI driver will continue to place max 10 shares per instance, irrespective of the actual max shares the Filestore instance can support.
+			// Alternately, if CSI max share features is enabled, but filestore disables the feature, the create volume may continue to fail beyond 10 shares per instance.
+			maxShareCount := util.MaxSharesPerInstance
+			if m.msControllerServer != nil && m.msControllerServer.featureMaxSharePerInstance {
+				maxShareCount = instance.MaxShareCount
+			}
+			if len(shares) >= maxShareCount {
 				continue
 			}
 


Test patch:
diff --git a/pkg/csi_driver/multishare_controller_test.go b/pkg/csi_driver/multishare_controller_test.go
index 0e1bf909..552e382d 100644
--- a/pkg/csi_driver/multishare_controller_test.go
+++ b/pkg/csi_driver/multishare_controller_test.go
@@ -21,6 +21,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"reflect"
+	"strconv"
 	"strings"
 	"testing"
 
@@ -66,6 +67,29 @@ func initTestMultishareController(t *testing.T) *MultishareController {
 	return NewMultishareController(config)
 }
 
+func initTestMultishareControllerWithFeatureOpts(t *testing.T, features *GCFSDriverFeatureOptions) *MultishareController {
+	fileService, err := file.NewFakeService()
+	if err != nil {
+		t.Fatalf("failed to initialize GCFS service: %v", err)
+	}
+
+	cloudProvider, err := cloud.NewFakeCloud()
+	if err != nil {
+		t.Fatalf("Failed to get cloud provider: %v", err)
+	}
+	config := &controllerServerConfig{
+		driver:          initTestDriver(t),
+		fileService:     fileService,
+		cloud:           cloudProvider,
+		volumeLocks:     util.NewVolumeLocks(),
+		ecfsDescription: "",
+		isRegional:      true,
+		clusterName:     testClusterName,
+		features:        features,
+	}
+	return NewMultishareController(config)
+}
+
 func TestPickRegion(t *testing.T) {
 	tests := []struct {
 		name           string
@@ -231,24 +255,47 @@ func TestGetShareRequestCapacity(t *testing.T) {
 	tests := []struct {
 		name             string
 		cap              *csi.CapacityRange
+		minSizeBytes     int64
+		maxSizeBytes     int64
 		expectErr        bool
 		expectedCapacity int64
 	}{
 		{
-			name:      "req and limit not set",
-			cap:       &csi.CapacityRange{},
-			expectErr: true,
+			name:         "req and limit not set",
+			cap:          &csi.CapacityRange{},
+			expectErr:    true,
+			minSizeBytes: util.MinShareSizeBytes,
+			maxSizeBytes: util.MaxShareSizeBytes,
 		},
 		{
-			name:             "empty cap range",
+			name:             "empty cap range, test 1",
 			expectedCapacity: 100 * util.Gb,
+			minSizeBytes:     util.MinShareSizeBytes,
+			maxSizeBytes:     util.MaxShareSizeBytes,
+		},
+		{
+			name:             "empty cap range, test 2",
+			expectedCapacity: util.MinShareSizeConfigurableBytes,
+			minSizeBytes:     util.MinShareSizeConfigurableBytes,
+			maxSizeBytes:     util.MaxShareSizeBytes,
 		},
 		{
-			name:      "cap range limit less than minimum supported size",
+			name:      "cap range limit less than minimum supported size, test 1",
 			expectErr: true,
 			cap: &csi.CapacityRange{
 				LimitBytes: 99 * util.Gb,
 			},
+			minSizeBytes: util.MinShareSizeBytes,
+			maxSizeBytes: util.MaxShareSizeBytes,
+		},
+		{
+			name:      "cap range limit less than minimum supported size, test 2",
+			expectErr: true,
+			cap: &csi.CapacityRange{
+				LimitBytes: 9 * util.Gb,
+			},
+			minSizeBytes: util.MinShareSizeConfigurableBytes,
+			maxSizeBytes: util.MaxShareSizeBytes,
 		},
 		{
 			name:      "cap range limit more than maximum supported size",
@@ -256,13 +303,26 @@ func TestGetShareRequestCapacity(t *testing.T) {
 			cap: &csi.CapacityRange{
 				LimitBytes: 2 * util.Tb,
 			},
+			minSizeBytes: util.MinShareSizeBytes,
+			maxSizeBytes: util.MaxShareSizeBytes,
 		},
 		{
-			name:      "cap range req less than minimum supported size",
+			name:      "cap range req less than minimum supported size, test 1",
 			expectErr: true,
 			cap: &csi.CapacityRange{
 				RequiredBytes: 99 * util.Gb,
 			},
+			minSizeBytes: util.MinShareSizeBytes,
+			maxSizeBytes: util.MaxShareSizeBytes,
+		},
+		{
+			name:      "cap range req less than minimum supported size, test 2",
+			expectErr: true,
+			cap: &csi.CapacityRange{
+				RequiredBytes: 9 * util.Gb,
+			},
+			minSizeBytes: util.MinShareSizeConfigurableBytes,
+			maxSizeBytes: util.MaxShareSizeBytes,
 		},
 		{
 			name:      "cap range req more than maximum supported size",
@@ -286,11 +346,62 @@ func TestGetShareRequestCapacity(t *testing.T) {
 				LimitBytes:    1 * util.Tb,
 			},
 			expectedCapacity: 1 * util.Tb,
+			minSizeBytes:     util.MinShareSizeBytes,
+			maxSizeBytes:     util.MaxShareSizeBytes,
+		},
+		{
+			name: "cap range req and limit set, limit exceeds min range",
+			cap: &csi.CapacityRange{
+				RequiredBytes: 1 * util.Gb,
+				LimitBytes:    9 * util.Gb,
+			},
+			expectErr:    true,
+			minSizeBytes: util.MinShareSizeConfigurableBytes,
+			maxSizeBytes: util.MaxShareSizeBytes,
+		},
+		{
+			name: "cap range req and limit set, limit exceeds min range",
+			cap: &csi.CapacityRange{
+				RequiredBytes: 1 * util.Gb,
+				LimitBytes:    99 * util.Gb,
+			},
+			expectErr:    true,
+			minSizeBytes: util.MinShareSizeBytes,
+			maxSizeBytes: util.MaxShareSizeBytes,
+		},
+		{
+			name: "cap range req and limit set within range",
+			cap: &csi.CapacityRange{
+				RequiredBytes: 100 * util.Gb,
+				LimitBytes:    100 * util.Gb,
+			},
+			minSizeBytes:     util.MinShareSizeConfigurableBytes,
+			maxSizeBytes:     128 * util.Gb,
+			expectedCapacity: 100 * util.Gb,
+		},
+		{
+			name: "cap range req and limit set, limit exceed range",
+			cap: &csi.CapacityRange{
+				RequiredBytes: 100 * util.Gb,
+				LimitBytes:    130 * util.Gb,
+			},
+			minSizeBytes: util.MinShareSizeConfigurableBytes,
+			maxSizeBytes: 128 * util.Gb,
+			expectErr:    true,
+		},
+		{
+			name: "cap range req set, req exceed range",
+			cap: &csi.CapacityRange{
+				RequiredBytes: 130 * util.Gb,
+			},
+			minSizeBytes: util.MinShareSizeConfigurableBytes,
+			maxSizeBytes: 128 * util.Gb,
+			expectErr:    true,
 		},
 	}
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			capacity, err := getShareRequestCapacity(tc.cap)
+			capacity, err := getShareRequestCapacity(tc.cap, tc.minSizeBytes, tc.maxSizeBytes)
 			if tc.expectErr && err == nil {
 				t.Error("expected error, got none")
 			}
@@ -478,7 +589,7 @@ func TestGenerateNewMultishareInstance(t *testing.T) {
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
 			m := initTestMultishareController(t)
-			filer, err := m.generateNewMultishareInstance(tc.instanceName, tc.req)
+			filer, err := m.generateNewMultishareInstance(tc.instanceName, tc.req, 10)
 			if tc.expectErr && err == nil {
 				t.Error("expected error, got none")
 			}
@@ -494,11 +605,13 @@ func TestGenerateNewMultishareInstance(t *testing.T) {
 
 func TestGenerateCSICreateVolumeResponse(t *testing.T) {
 	tests := []struct {
-		name         string
-		prefix       string
-		share        *file.Share
-		expectedResp *csi.CreateVolumeResponse
-		expectError  bool
+		name              string
+		prefix            string
+		share             *file.Share
+		expectError       bool
+		features          *GCFSDriverFeatureOptions
+		expectedResp      *csi.CreateVolumeResponse
+		maxShareSizeBytes int64
 	}{
 		{
 			name:        "empty prefix",
@@ -532,6 +645,7 @@ func TestGenerateCSICreateVolumeResponse(t *testing.T) {
 				},
 				CapacityBytes: 1 * util.Tb,
 			},
+			maxShareSizeBytes: 1 * util.Tb,
 			expectedResp: &csi.CreateVolumeResponse{
 				Volume: &csi.Volume{
 					VolumeId:      modeMultishare + "/" + testInstanceScPrefix + "/" + testProject + "/" + testLocation + "/" + testInstanceName + "/" + testShareName,
@@ -542,11 +656,76 @@ func TestGenerateCSICreateVolumeResponse(t *testing.T) {
 				},
 			},
 		},
+		{
+			name:   "valid share object, with configurable share feature enabled",
+			prefix: testInstanceScPrefix,
+			features: &GCFSDriverFeatureOptions{
+				FeatureMaxSharesPerInstance: &FeatureMaxSharesPerInstance{
+					Enabled: true,
+				},
+			},
+			share: &file.Share{
+				Name: testShareName,
+				Parent: &file.MultishareInstance{
+					Name:     testInstanceName,
+					Project:  testProject,
+					Location: testLocation,
+					Network: file.Network{
+						Ip: "1.1.1.1",
+					},
+				},
+				CapacityBytes: 1 * util.Tb,
+			},
+			maxShareSizeBytes: 1 * util.Tb,
+			expectedResp: &csi.CreateVolumeResponse{
+				Volume: &csi.Volume{
+					VolumeId:      modeMultishare + "/" + testInstanceScPrefix + "/" + testProject + "/" + testLocation + "/" + testInstanceName + "/" + testShareName,
+					CapacityBytes: 1 * util.Tb,
+					VolumeContext: map[string]string{
+						attrIP:           "1.1.1.1",
+						attrMaxShareSize: strconv.Itoa(util.Tb),
+					},
+				},
+			},
+		},
+		{
+			name:   "valid share object, with configurable share feature enabled, test 2",
+			prefix: testInstanceScPrefix,
+			features: &GCFSDriverFeatureOptions{
+				FeatureMaxSharesPerInstance: &FeatureMaxSharesPerInstance{
+					Enabled: true,
+				},
+			},
+			share: &file.Share{
+				Name: testShareName,
+				Parent: &file.MultishareInstance{
+					Name:     testInstanceName,
+					Project:  testProject,
+					Location: testLocation,
+					Network: file.Network{
+						Ip: "1.1.1.1",
+					},
+				},
+				CapacityBytes: 1 * util.Tb,
+			},
+			maxShareSizeBytes: 100 * util.Gb,
+			expectedResp: &csi.CreateVolumeResponse{
+				Volume: &csi.Volume{
+					VolumeId:      modeMultishare + "/" + testInstanceScPrefix + "/" + testProject + "/" + testLocation + "/" + testInstanceName + "/" + testShareName,
+					CapacityBytes: 1 * util.Tb,
+					VolumeContext: map[string]string{
+						attrIP:           "1.1.1.1",
+						attrMaxShareSize: strconv.Itoa(100 * util.Gb),
+					},
+				},
+			},
+		},
 	}
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			m := initTestMultishareController(t)
-			resp, err := m.generateCSICreateVolumeResponse(tc.prefix, tc.share)
+			//m := initTestMultishareController(t)
+			m := initTestMultishareControllerWithFeatureOpts(t, tc.features)
+			resp, err := m.generateCSICreateVolumeResponse(tc.prefix, tc.share, tc.maxShareSizeBytes)
 			if tc.expectError && err == nil {
 				t.Error("expected error, got none")
 			}
@@ -1621,3 +1800,251 @@ func TestMultishareControllerExpandVolume(t *testing.T) {
 		})
 	}
 }
+
+func TestIsValidMaxVolSize(t *testing.T) {
+	tests := []struct {
+		val         int64
+		expectedret bool
+	}{
+		{
+			val: 0,
+		},
+		{
+			val:         128 * util.Gb,
+			expectedret: true,
+		},
+		{
+			val:         256 * util.Gb,
+			expectedret: true,
+		},
+		{
+			val:         512 * util.Gb,
+			expectedret: true,
+		},
+		{
+			val:         1 * util.Tb,
+			expectedret: true,
+		},
+		{
+			val:         1024 * util.Gb,
+			expectedret: true,
+		},
+		{
+			val:         131072 * util.Mb,
+			expectedret: true,
+		},
+		// Negative test cases
+		{
+			val:         131073 * util.Mb,
+			expectedret: false,
+		},
+		{
+			val:         131073,
+			expectedret: false,
+		},
+		{
+			val:         1023 * util.Tb,
+			expectedret: false,
+		},
+		{
+			val:         127 * util.Gb,
+			expectedret: false,
+		},
+		{
+			val:         257 * util.Gb,
+			expectedret: false,
+		},
+		{
+			val:         -1024 * util.Gb,
+			expectedret: false,
+		},
+		{
+			val:         0,
+			expectedret: false,
+		},
+	}
+	for _, tc := range tests {
+		if isValidMaxVolSize(tc.val) != tc.expectedret {
+			t.Errorf("failed for val %d, expected ret %v", tc.val, tc.expectedret)
+		}
+	}
+}
+
+func TestParseMaxVolumeSizeParam(t *testing.T) {
+	tests := []struct {
+		name                          string
+		req                           *csi.CreateVolumeRequest
+		features                      *GCFSDriverFeatureOptions
+		expectedSharesPerInstance     int
+		expectedMaxShareCapacityBytes int64
+		expectError                   bool
+	}{
+		{
+			name: "feature not enabled, param set",
+			req: &csi.CreateVolumeRequest{
+				Parameters: map[string]string{
+					paramMaxVolumeSize: "128Gi",
+				},
+			},
+			expectError: true,
+		},
+		{
+			name: "feature enabled, param set, value not set",
+			features: &GCFSDriverFeatureOptions{
+				FeatureMaxSharesPerInstance: &FeatureMaxSharesPerInstance{
+					Enabled: true,
+				},
+			},
+			req: &csi.CreateVolumeRequest{
+				Parameters: map[string]string{
+					paramMaxVolumeSize: "",
+				},
+			},
+			expectError: true,
+		},
+		{
+			name: "feature enabled, param set, invalid value test1",
+			features: &GCFSDriverFeatureOptions{
+				FeatureMaxSharesPerInstance: &FeatureMaxSharesPerInstance{
+					Enabled: true,
+				},
+			},
+			req: &csi.CreateVolumeRequest{
+				Parameters: map[string]string{
+					paramMaxVolumeSize: "-128Gi",
+				},
+			},
+			expectError: true,
+		},
+		{
+			name: "feature enabled, param set, invalid value",
+			features: &GCFSDriverFeatureOptions{
+				FeatureMaxSharesPerInstance: &FeatureMaxSharesPerInstance{
+					Enabled: true,
+				},
+			},
+			req: &csi.CreateVolumeRequest{
+				Parameters: map[string]string{
+					paramMaxVolumeSize: "12i",
+				},
+			},
+			expectError: true,
+		},
+		{
+			name: "feature enabled, param set, unexpected value test1",
+			features: &GCFSDriverFeatureOptions{
+				FeatureMaxSharesPerInstance: &FeatureMaxSharesPerInstance{
+					Enabled: true,
+				},
+			},
+			req: &csi.CreateVolumeRequest{
+				Parameters: map[string]string{
+					paramMaxVolumeSize: "100Gi",
+				},
+			},
+			expectError: true,
+		},
+		{
+			name: "feature enabled, param set, unexpected value test2",
+			features: &GCFSDriverFeatureOptions{
+				FeatureMaxSharesPerInstance: &FeatureMaxSharesPerInstance{
+					Enabled: true,
+				},
+			},
+			req: &csi.CreateVolumeRequest{
+				Parameters: map[string]string{
+					paramMaxVolumeSize: "100Gi",
+				},
+			},
+			expectError: true,
+		},
+		{
+			name: "feature enabled, param not set, defaults returned",
+			features: &GCFSDriverFeatureOptions{
+				FeatureMaxSharesPerInstance: &FeatureMaxSharesPerInstance{
+					Enabled: true,
+				},
+			},
+			req: &csi.CreateVolumeRequest{
+				Parameters: map[string]string{},
+			},
+			expectedSharesPerInstance:     util.MaxSharesPerInstance,
+			expectedMaxShareCapacityBytes: util.MaxShareSizeBytes,
+		},
+		{
+			name: "feature enabled, param set success test1",
+			features: &GCFSDriverFeatureOptions{
+				FeatureMaxSharesPerInstance: &FeatureMaxSharesPerInstance{
+					Enabled: true,
+				},
+			},
+			req: &csi.CreateVolumeRequest{
+				Parameters: map[string]string{
+					paramMaxVolumeSize: "128Gi",
+				},
+			},
+			expectedSharesPerInstance:     80,
+			expectedMaxShareCapacityBytes: 128 * util.Gb,
+		},
+		{
+			name: "feature enabled, param set success test2",
+			features: &GCFSDriverFeatureOptions{
+				FeatureMaxSharesPerInstance: &FeatureMaxSharesPerInstance{
+					Enabled: true,
+				},
+			},
+			req: &csi.CreateVolumeRequest{
+				Parameters: map[string]string{
+					paramMaxVolumeSize: "256Gi",
+				},
+			},
+			expectedSharesPerInstance:     40,
+			expectedMaxShareCapacityBytes: 256 * util.Gb,
+		},
+		{
+			name: "feature enabled, param set success test3",
+			features: &GCFSDriverFeatureOptions{
+				FeatureMaxSharesPerInstance: &FeatureMaxSharesPerInstance{
+					Enabled: true,
+				},
+			},
+			req: &csi.CreateVolumeRequest{
+				Parameters: map[string]string{
+					paramMaxVolumeSize: "512Gi",
+				},
+			},
+			expectedSharesPerInstance:     20,
+			expectedMaxShareCapacityBytes: 512 * util.Gb,
+		},
+		{
+			name: "feature enabled, param set success test4",
+			features: &GCFSDriverFeatureOptions{
+				FeatureMaxSharesPerInstance: &FeatureMaxSharesPerInstance{
+					Enabled: true,
+				},
+			},
+			req: &csi.CreateVolumeRequest{
+				Parameters: map[string]string{
+					paramMaxVolumeSize: "1024Gi",
+				},
+			},
+			expectedSharesPerInstance:     10,
+			expectedMaxShareCapacityBytes: 1 * util.Tb,
+		},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			m := initTestMultishareControllerWithFeatureOpts(t, tc.features)
+			sharePerInstance, maxShareCapacity, err := m.parseMaxVolumeSizeParam(tc.req)
+			if tc.expectError && err == nil {
+				t.Errorf("failed")
+			}
+			if !tc.expectError && err != nil {
+				t.Errorf("failed")
+			}
+			if sharePerInstance != tc.expectedSharesPerInstance || maxShareCapacity != tc.expectedMaxShareCapacityBytes {
+				t.Errorf("failed")
+			}
+		})
+	}
+}
diff --git a/pkg/csi_driver/multishare_ops_manager_test.go b/pkg/csi_driver/multishare_ops_manager_test.go
index 7325135a..b8dcb7f4 100644
--- a/pkg/csi_driver/multishare_ops_manager_test.go
+++ b/pkg/csi_driver/multishare_ops_manager_test.go
@@ -1634,6 +1634,7 @@ func TestRunEligibleInstanceCheck(t *testing.T) {
 		req                   *csi.CreateVolumeRequest
 		target                *file.MultishareInstance
 		expectError           bool
+		features              *GCFSDriverFeatureOptions
 	}{
 		{
 			name: "no instances",
@@ -1668,7 +1669,7 @@ func TestRunEligibleInstanceCheck(t *testing.T) {
 					State: "READY",
 				},
 				{
-					Name:     "test-instance-1",
+					Name:     "test-instance-2",
 					Project:  testProject,
 					Location: testRegion,
 					Labels: map[string]string{
@@ -2290,6 +2291,175 @@ func TestRunEligibleInstanceCheck(t *testing.T) {
 				},
 			},
 		},
+		{
+			name: "instance exhausted with max shares, no ready instance found",
+			features: &GCFSDriverFeatureOptions{
+				FeatureMaxSharesPerInstance: &FeatureMaxSharesPerInstance{
+					Enabled: true,
+				},
+			},
+			req: &csi.CreateVolumeRequest{
+				Parameters: map[string]string{
+					paramMultishareInstanceScLabel: testInstanceScPrefix,
+				},
+			},
+			target: &file.MultishareInstance{
+				Name:     "test-target-instance",
+				Project:  testProject,
+				Location: testRegion,
+				Labels: map[string]string{
+					util.ParamMultishareInstanceScLabelKey: testInstanceScPrefix,
+					tagKeyClusterLocation:                  testLocation,
+					tagKeyClusterName:                      testClusterName,
+				},
+			},
+			initInstances: []*file.MultishareInstance{
+				{
+					Name:     "instance-1",
+					Project:  testProject,
+					Location: testRegion,
+					Labels: map[string]string{
+						util.ParamMultishareInstanceScLabelKey: testInstanceScPrefix,
+						tagKeyClusterLocation:                  testLocation,
+						tagKeyClusterName:                      testClusterName,
+					},
+					State:         "READY",
+					MaxShareCount: 2,
+				},
+			},
+			initShares: []*file.Share{
+				{
+					Name: "share-1",
+					Parent: &file.MultishareInstance{
+						Name:     "instance-1",
+						Project:  testProject,
+						Location: testRegion,
+						Labels: map[string]string{
+							util.ParamMultishareInstanceScLabelKey: testInstanceScPrefix,
+							tagKeyClusterLocation:                  testLocation,
+							tagKeyClusterName:                      testClusterName,
+						},
+					},
+				},
+				{
+					Name: "share-2",
+					Parent: &file.MultishareInstance{
+						Name:     "instance-1",
+						Project:  testProject,
+						Location: testRegion,
+						Labels: map[string]string{
+							util.ParamMultishareInstanceScLabelKey: testInstanceScPrefix,
+							tagKeyClusterLocation:                  testLocation,
+							tagKeyClusterName:                      testClusterName,
+						},
+					},
+				},
+			},
+		},
+		{
+			name: "1 instance exhausted with max shares, 1 ready instance with less than max share count",
+			features: &GCFSDriverFeatureOptions{
+				FeatureMaxSharesPerInstance: &FeatureMaxSharesPerInstance{
+					Enabled: true,
+				},
+			},
+			req: &csi.CreateVolumeRequest{
+				Parameters: map[string]string{
+					paramMultishareInstanceScLabel: testInstanceScPrefix,
+				},
+			},
+			target: &file.MultishareInstance{
+				Name:     "test-target-instance",
+				Project:  testProject,
+				Location: testRegion,
+				Labels: map[string]string{
+					util.ParamMultishareInstanceScLabelKey: testInstanceScPrefix,
+					tagKeyClusterLocation:                  testLocation,
+					tagKeyClusterName:                      testClusterName,
+				},
+			},
+			initInstances: []*file.MultishareInstance{
+				{
+					Name:     "instance-1",
+					Project:  testProject,
+					Location: testRegion,
+					Labels: map[string]string{
+						util.ParamMultishareInstanceScLabelKey: testInstanceScPrefix,
+						tagKeyClusterLocation:                  testLocation,
+						tagKeyClusterName:                      testClusterName,
+					},
+					State:         "READY",
+					MaxShareCount: 2,
+				},
+				{
+					Name:     "instance-2",
+					Project:  testProject,
+					Location: testRegion,
+					Labels: map[string]string{
+						util.ParamMultishareInstanceScLabelKey: testInstanceScPrefix,
+						tagKeyClusterLocation:                  testLocation,
+						tagKeyClusterName:                      testClusterName,
+					},
+					State:         "READY",
+					MaxShareCount: 10,
+				},
+			},
+			initShares: []*file.Share{
+				{
+					Name: "share-1",
+					Parent: &file.MultishareInstance{
+						Name:     "instance-1",
+						Project:  testProject,
+						Location: testRegion,
+						Labels: map[string]string{
+							util.ParamMultishareInstanceScLabelKey: testInstanceScPrefix,
+							tagKeyClusterLocation:                  testLocation,
+							tagKeyClusterName:                      testClusterName,
+						},
+					},
+				},
+				{
+					Name: "share-2",
+					Parent: &file.MultishareInstance{
+						Name:     "instance-1",
+						Project:  testProject,
+						Location: testRegion,
+						Labels: map[string]string{
+							util.ParamMultishareInstanceScLabelKey: testInstanceScPrefix,
+							tagKeyClusterLocation:                  testLocation,
+							tagKeyClusterName:                      testClusterName,
+						},
+					},
+				},
+				{
+					Name: "share-3",
+					Parent: &file.MultishareInstance{
+						Name:     "instance-2",
+						Project:  testProject,
+						Location: testRegion,
+						Labels: map[string]string{
+							util.ParamMultishareInstanceScLabelKey: testInstanceScPrefix,
+							tagKeyClusterLocation:                  testLocation,
+							tagKeyClusterName:                      testClusterName,
+						},
+					},
+				},
+			},
+			expectedReadyInstance: []*file.MultishareInstance{
+				{
+					Name:     "instance-2",
+					Project:  testProject,
+					Location: testRegion,
+					Labels: map[string]string{
+						util.ParamMultishareInstanceScLabelKey: testInstanceScPrefix,
+						tagKeyClusterLocation:                  testLocation,
+						tagKeyClusterName:                      testClusterName,
+					},
+					State:         "READY",
+					MaxShareCount: 10,
+				},
+			},
+		},
 	}
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
@@ -2303,6 +2473,7 @@ func TestRunEligibleInstanceCheck(t *testing.T) {
 				driver:      initTestDriver(t),
 				fileService: s,
 				cloud:       cloudProvider,
+				features:    tc.features,
 			}
 			mcs := NewMultishareController(config)
 			ready, err := mcs.opsManager.runEligibleInstanceCheck(context.Background(), tc.req, tc.ops, tc.target, testRegions)
@@ -2313,6 +2484,9 @@ func TestRunEligibleInstanceCheck(t *testing.T) {
 			if tc.expectError && err == nil {
 				t.Errorf("expected error")
 			}
+			if len(ready) != len(tc.expectedReadyInstance) {
+				t.Errorf("Mismatch in expected ready instances count, ready %d, expected %d", len(ready), len(tc.expectedReadyInstance))
+			}
 			for _, r := range ready {
 				if !found(tc.expectedReadyInstance, r) {
 					t.Errorf("expected instance not ready")