Import changes.

  - 2d2e88908cd1911ca29161030dc47078cd8f61f6
  - 090224466e5778e6524feb17ad5bc05492410583

GitOrigin-RevId: 2d2e88908cd1911ca29161030dc47078cd8f61f6

Author: Aalyria Technologies, Inc

tools/nbictl/README.md

@@ -135,6 +135,8 @@ Sync all model entities and relationships from file and directory arguments.
 
 **--format**="": The format to use for encoding and decoding protobuf messages. One of [text, json, binary]. (default: "text")
 
+**--max-concurrency, -j, --max_concurrency**="": Limit the number of in-flight requests at once. (default: 100)
+
 **--recursive, -r**: descend recursively into directory arguments
 
 **--verbose, -v**: increase verbosity
@@ -161,6 +163,8 @@ Sync all provisioning resources from file and directory arguments.
 
 **--format**="": The format to use for encoding and decoding protobuf messages. One of [text, json, binary]. (default: "text")
 
+**--max-concurrency, -j, --max_concurrency**="": Limit the number of in-flight requests at once. (default: 100)
+
 **--recursive, -r**: descend recursively into directory arguments
 
 **--verbose, -v**: increase verbosity

tools/nbictl/model.go

@@ -445,6 +445,7 @@ func ModelSync(appCtx *cli.Context) error {
 	deleteMode := appCtx.Bool("delete")
 	dryRunMode := appCtx.Bool("dry-run")
 	verboseMode := appCtx.Bool("verbose")
+	maxConcurrency := appCtx.Int("max-concurrency")
 	printMode := dryRunMode || verboseMode
 
 	marshaller, err := marshallerForFormat(appCtx.String("format"))
@@ -545,7 +546,7 @@ func ModelSync(appCtx *cli.Context) error {
 	errs := []error{}
 
 	// Delete relationships before deleting entities.
-	deleteRelationshipsPool := pool.New().WithErrors()
+	deleteRelationshipsPool := pool.New().WithErrors().WithMaxGoroutines(maxConcurrency)
 	if deleteMode {
 		relationshipsToBeDeleted := remoteRelationshipKeys.Difference(localRelationshipKeys)
 		for relationship := range relationshipsToBeDeleted.Iter() {
@@ -568,7 +569,7 @@ func ModelSync(appCtx *cli.Context) error {
 	errs = append(errs, deleteRelationshipsPool.Wait())
 
 	// Update entities.
-	updatePool := pool.New().WithErrors()
+	updatePool := pool.New().WithErrors().WithMaxGoroutines(maxConcurrency)
 	entitiesInCommon := localEntityKeys.Intersect(remoteEntityKeys)
 	for entity := range entitiesInCommon.Iter() {
 		updatePool.Go(func() error {
@@ -589,12 +590,11 @@ func ModelSync(appCtx *cli.Context) error {
 	}
 
 	// Maybe delete entities (and prune collaterally deleted relationships)
-	deleteEntitiesPool := pool.New().WithErrors()
 	if deleteMode {
 		entitiesToBeDeleted := remoteEntityKeys.Difference(localEntityKeys)
 
 		for entity := range entitiesToBeDeleted.Iter() {
-			deleteEntitiesPool.Go(func() error {
+			updatePool.Go(func() error {
 				if printMode {
 					fmt.Printf("delete entity: %s\n", entity)
 				}
@@ -613,10 +613,9 @@ func ModelSync(appCtx *cli.Context) error {
 	}
 
 	errs = append(errs, updatePool.Wait())
-	errs = append(errs, deleteEntitiesPool.Wait())
 
 	// Add entities.
-	addEntitiesPool := pool.New().WithErrors()
+	addEntitiesPool := pool.New().WithErrors().WithMaxGoroutines(maxConcurrency)
 	entitiesToBeAdded := localEntityKeys.Difference(remoteEntityKeys)
 	for entity := range entitiesToBeAdded.Iter() {
 		addEntitiesPool.Go(func() error {
@@ -637,7 +636,7 @@ func ModelSync(appCtx *cli.Context) error {
 	errs = append(errs, addEntitiesPool.Wait())
 
 	// Add relationships.
-	addRelationshipsPool := pool.New().WithErrors()
+	addRelationshipsPool := pool.New().WithErrors().WithMaxGoroutines(maxConcurrency)
 	relationshipsToBeAdded := localRelationshipKeys.Difference(remoteRelationshipKeys)
 	for relationship := range relationshipsToBeAdded.Iter() {
 		addRelationshipsPool.Go(func() error {

tools/nbictl/nbictl.go

@@ -102,6 +102,12 @@ func App() *cli.App {
 		Usage:   "increase verbosity",
 		Aliases: []string{"v"},
 	}
+	concurrencyFlag := &cli.IntFlag{
+		Name:    "max-concurrency",
+		Aliases: []string{"j", "max_concurrency"},
+		Value:   100,
+		Usage:   "Limit the number of in-flight requests at once.",
+	}
 
 	return &cli.App{
 		Name:                 appName,
@@ -315,6 +321,7 @@ func App() *cli.App {
 							formatFlag,
 							dryrunFlag,
 							verboseFlag,
+							concurrencyFlag,
 							&cli.BoolFlag{
 								Name:    "delete",
 								Usage:   "delete entities and relationships from remote instance not present in local sources",
@@ -354,6 +361,7 @@ func App() *cli.App {
 							formatFlag,
 							dryrunFlag,
 							verboseFlag,
+							concurrencyFlag,
 							&cli.BoolFlag{
 								Name:    "delete",
 								Usage:   "delete resources from remote instance not present in local sources",

tools/nbictl/provisioning.go

@@ -436,6 +436,7 @@ func ProvisioningSync(appCtx *cli.Context) error {
 	deleteMode := appCtx.Bool("delete")
 	dryRunMode := appCtx.Bool("dry-run")
 	verboseMode := appCtx.Bool("verbose")
+	maxConcurrency := appCtx.Int("max-concurrency")
 	printMode := dryRunMode || verboseMode
 
 	// TODO: restructure these computations for scalability.
@@ -504,7 +505,7 @@ func ProvisioningSync(appCtx *cli.Context) error {
 		errs = append(errs, err)
 	}
 
-	p := pool.New().WithErrors()
+	p := pool.New().WithErrors().WithMaxGoroutines(maxConcurrency)
 
 	///
 	// Update resources.

version.bzl

@@ -25,7 +25,7 @@ MAJOR = 19
 # -------------------------------------------------------------
 # This allows to publish bugfixes to the released version
 # through increments of PATCH segment.
-MINOR = 10
+MINOR = 11
 
 # Monotonic ID of bugfixes applied to each MAJOR.MINOR release.
 # -------------------------------------------------------------