Add --disable-concurrency-limiter cli flag (#3203)

<!-- PR description-->
The concurrency limiter middleware is enabled by default for exchange backups. Adding a hidden --disable-concurrency-limiter flag to disable this middleware.

Reasons for adding this flag:
We have done limited perf testing with the concurrency limiter. In addition, our understanding of exchange concurrency limits is also a bit fuzzy. This flag acts as a killswitch in case we start to see perf regressions in prod.

I see this as a temporary flag. Ideally we would never have to use it. Also, once we have a better way to configure concurrency limiter, we can eliminate this flag.


---

#### Does this PR need a docs update or release note?

- [ ] ⛔ No

#### Type of change

<!--- Please check the type of change your PR introduces: --->
- [ ] 🌻 Feature


#### Issue(s)

<!-- Can reference multiple issues. Use one of the following "magic words" - "closes, fixes" to auto-close the Github issue. -->
* #<issue>

#### Test Plan

<!-- How will this be tested prior to merging.-->

- [ ] ⚡ Unit test

src/cli/backup/exchange.go

@@ -89,6 +89,7 @@ func addExchangeCommands(cmd *cobra.Command) *cobra.Command {
 		options.AddFailFastFlag(c)
 		options.AddDisableIncrementalsFlag(c)
 		options.AddEnableImmutableIDFlag(c)
+		options.AddDisableConcurrencyLimiterFlag(c)
 
 	case listCommand:
 		c, fs = utils.AddCommand(cmd, exchangeListCmd())

src/cli/options/options.go

@@ -19,6 +19,7 @@ func Control() control.Options {
 	opt.SkipReduce = skipReduceFV
 	opt.ToggleFeatures.DisableIncrementals = disableIncrementalsFV
 	opt.ToggleFeatures.ExchangeImmutableIDs = enableImmutableID
+	opt.ToggleFeatures.DisableConcurrencyLimiter = disableConcurrencyLimiterFV
 	opt.Parallelism.ItemFetch = fetchParallelismFV
 
 	return opt
@@ -29,13 +30,14 @@ func Control() control.Options {
 // ---------------------------------------------------------------------------
 
 const (
-	FailFastFN            = "fail-fast"
+	FailFastFN                  = "fail-fast"
-	FetchParallelismFN    = "fetch-parallelism"
+	FetchParallelismFN          = "fetch-parallelism"
-	NoStatsFN             = "no-stats"
+	NoStatsFN                   = "no-stats"
-	RestorePermissionsFN  = "restore-permissions"
+	RestorePermissionsFN        = "restore-permissions"
-	SkipReduceFN          = "skip-reduce"
+	SkipReduceFN                = "skip-reduce"
-	DisableIncrementalsFN = "disable-incrementals"
+	DisableIncrementalsFN       = "disable-incrementals"
-	EnableImmutableIDFN   = "enable-immutable-id"
+	EnableImmutableIDFN         = "enable-immutable-id"
+	DisableConcurrencyLimiterFN = "disable-concurrency-limiter"
 )
 
 var (
@@ -117,3 +119,18 @@ func AddEnableImmutableIDFlag(cmd *cobra.Command) {
 		"Enable exchange immutable ID.")
 	cobra.CheckErr(fs.MarkHidden(EnableImmutableIDFN))
 }
+
+var disableConcurrencyLimiterFV bool
+
+// AddDisableConcurrencyLimiterFlag adds a hidden cli flag which, when set,
+// removes concurrency limits when communicating with graph API. This
+// flag is only relevant for exchange backups for now
+func AddDisableConcurrencyLimiterFlag(cmd *cobra.Command) {
+	fs := cmd.Flags()
+	fs.BoolVar(
+		&disableConcurrencyLimiterFV,
+		DisableConcurrencyLimiterFN,
+		false,
+		"Disable concurrency limiter middleware. Default: false")
+	cobra.CheckErr(fs.MarkHidden(DisableConcurrencyLimiterFN))
+}

src/cli/options/options_test.go

@@ -32,6 +32,7 @@ func (suite *OptionsUnitSuite) TestAddExchangeCommands() {
 			assert.True(t, restorePermissionsFV, RestorePermissionsFN)
 			assert.True(t, skipReduceFV, SkipReduceFN)
 			assert.Equal(t, 2, fetchParallelismFV, FetchParallelismFN)
+			assert.True(t, disableConcurrencyLimiterFV, DisableConcurrencyLimiterFN)
 		},
 	}
 
@@ -42,8 +43,8 @@ func (suite *OptionsUnitSuite) TestAddExchangeCommands() {
 	AddDisableIncrementalsFlag(cmd)
 	AddRestorePermissionsFlag(cmd)
 	AddSkipReduceFlag(cmd)
-
 	AddFetchParallelismFlag(cmd)
+	AddDisableConcurrencyLimiterFlag(cmd)
 
 	// Test arg parsing for few args
 	cmd.SetArgs([]string{
@@ -53,8 +54,8 @@ func (suite *OptionsUnitSuite) TestAddExchangeCommands() {
 		"--" + NoStatsFN,
 		"--" + RestorePermissionsFN,
 		"--" + SkipReduceFN,
-
 		"--" + FetchParallelismFN, "2",
+		"--" + DisableConcurrencyLimiterFN,
 	})
 
 	err := cmd.Execute()

src/internal/connector/exchange/data_collections.go

@@ -182,8 +182,11 @@ func DataCollections(
 		categories  = map[path.CategoryType]struct{}{}
 	)
 
-	// TODO: Add hidden cli flag to disable this feature
+	// Turn on concurrency limiter middleware for exchange backups
-	graph.InitializeConcurrencyLimiter(ctrlOpts.Parallelism.ItemFetch)
+	// unless explicitly disabled through DisableConcurrencyLimiterFN cli flag
+	if !ctrlOpts.ToggleFeatures.DisableConcurrencyLimiter {
+		graph.InitializeConcurrencyLimiter(ctrlOpts.Parallelism.ItemFetch)
+	}
 
 	cdps, err := parseMetadataCollections(ctx, metadata, errs)
 	if err != nil {

src/pkg/control/options.go

@@ -103,4 +103,10 @@ type Toggles struct {
 	// immutable Exchange IDs. This is only safe to set if the previous backup for
 	// incremental backups used immutable IDs or if a full backup is being done.
 	ExchangeImmutableIDs bool `json:"exchangeImmutableIDs,omitempty"`
+
+	RunMigrations bool `json:"runMigrations"`
+
+	// DisableConcurrencyLimiter removes concurrency limits when communicating with
+	// graph API. This flag is only relevant for exchange backups for now
+	DisableConcurrencyLimiter bool `json:"disableConcurrencyLimiter,omitempty"`
 }