FS-1252; Streamline fleet-scheduler inventory task threading by removing threading (#36)

Author: jakeschuurmans

cmd/inventory.go

@@ -13,8 +13,7 @@ import (
 )
 
 var (
-	pageSize      int
-	inFlightPages int
+	pageSize int
 )
 
 var cmdInventory = &cobra.Command{
@@ -31,7 +30,6 @@ var cmdInventory = &cobra.Command{
 
 func init() {
 	rootCmd.PersistentFlags().IntVar(&pageSize, "page-size", 4, "Define how many servers to query per request")
-	rootCmd.PersistentFlags().IntVar(&inFlightPages, "inflight-pages", 1, "Define how many server pages to queue up before waiting for the previous to finish creating the condition")
 	rootCmd.AddCommand(cmdInventory)
 }
 
@@ -66,7 +64,7 @@ func inventory(ctx context.Context) error {
 		return err
 	}
 
-	err = newClient.CreateConditionInventoryForAllServers(pageSize, inFlightPages)
+	err = newClient.CreateConditionInventoryForAllServers(pageSize)
 	if err != nil {
 		return err
 	}

cmd/test.go

@@ -1,81 +1,10 @@
 package client
 
 import (
-	"time"
-
-	"github.com/sirupsen/logrus"
-	"golang.org/x/sync/semaphore"
-
 	fleetdbapi "github.com/metal-toolbox/fleetdb/pkg/api/v1"
 	fleetDBRivets "github.com/metal-toolbox/rivets/serverservice"
 )
 
-func (c *Client) gatherServers(pageSize int, serverCh chan *fleetdbapi.Server, concLimiter *semaphore.Weighted) {
-	// signal to receiver that we are done
-	defer close(serverCh)
-
-	// First page, use the response from it to figure out how many pages we have to loop through
-	// Dont change page size
-	servers, response, err := c.getServerPage(pageSize, 1)
-	if err != nil {
-		c.log.WithFields(logrus.Fields{
-			"pageSize":  pageSize,
-			"pageIndex": 1,
-		}).Logger.Errorf("Failed to get list of servers: %s", err.Error())
-		return
-	}
-	totalPages := response.TotalPages
-
-	if !concLimiter.TryAcquire(int64(response.PageSize)) {
-		c.log.Error("Failed to acquire semaphore! Going to attempt to continue.")
-	}
-
-	// send first page of servers to the channel
-	for i := range servers {
-		serverCh <- &servers[i]
-	}
-
-	c.log.WithFields(logrus.Fields{
-		"index":      1,
-		"iterations": totalPages,
-		"got":        len(servers),
-	}).Trace("Got server page")
-
-	// Start the second page, and loop through rest the pages
-	for i := 2; i <= totalPages; i++ {
-		servers, response, err = c.getServerPage(pageSize, i)
-		if err != nil {
-			c.log.WithFields(logrus.Fields{
-				"pageSize":  pageSize,
-				"pageIndex": i,
-			}).Logger.Errorf("Failed to get page of servers, attempting to continue: %s", err.Error())
-
-			continue
-		}
-
-		c.log.WithFields(logrus.Fields{
-			"index":      i,
-			"iterations": totalPages,
-			"got":        len(servers),
-		}).Trace("Got server page")
-
-		// throttle this loop
-		// Doing a spinlock to prevent a permanent lock if the ctx gets canceled
-		for !concLimiter.TryAcquire(int64(response.PageSize)) && c.ctx.Err() == nil {
-			time.Sleep(time.Second)
-		}
-
-		if c.ctx.Err() != nil {
-			c.log.Warn("Context canceled, stopping server gathering")
-			return
-		}
-
-		for i := range servers {
-			serverCh <- &servers[i]
-		}
-	}
-}
-
 func (c *Client) getServerPage(pageSize, page int) ([]fleetdbapi.Server, *fleetdbapi.ServerResponse, error) {
 	params := &fleetdbapi.ServerListParams{
 		FacilityCode: c.cfg.FacilityCode,

internal/client/fleetdb.go

@@ -2,41 +2,60 @@ package client
 
 import (
 	"github.com/sirupsen/logrus"
-	"golang.org/x/sync/semaphore"
-
-	// "github.com/sirupsen/logrus"
-	fleetdbapi "github.com/metal-toolbox/fleetdb/pkg/api/v1"
 )
 
-func (c *Client) CreateConditionInventoryForAllServers(pageSize, inFlightPages int) error {
-	// Start thread to start collecting servers
-	serverCh, concLimiter, err := c.GatherServersNonBlocking(pageSize, inFlightPages)
+func (c *Client) CreateConditionInventoryForAllServers(pageSize int) error {
+	// First page, use the response from it to figure out how many pages we have to loop through
+	// Dont change page size
+	servers, response, err := c.getServerPage(pageSize, 1)
 	if err != nil {
+		c.log.WithFields(logrus.Fields{
+			"pageSize":  pageSize,
+			"pageIndex": 1,
+		}).Logger.Errorf("Failed to get list of servers: %s", err.Error())
 		return err
 	}
+	totalPages := response.TotalPages
 
-	// Loop through servers and create conditions
-	for server := range serverCh {
-		err := c.CreateConditionInventory(server.UUID)
+	// send first page of servers to the channel
+	for i := range servers {
+		err = c.CreateConditionInventory(servers[i].UUID)
 		if err != nil {
-			c.log.WithFields(logrus.Fields{
-				"server": server.UUID,
-			}).Logger.Error("Failed to create condition")
+			return err
 		}
-
-		concLimiter.Release(1)
 	}
 
-	return nil
-}
+	c.log.WithFields(logrus.Fields{
+		"index":      1,
+		"iterations": totalPages,
+		"got":        len(servers),
+	}).Trace("Got server page")
+
+	// Start the second page, and loop through rest the pages
+	for i := 2; i <= totalPages; i++ {
+		servers, _, err = c.getServerPage(pageSize, i)
+		if err != nil {
+			c.log.WithFields(logrus.Fields{
+				"pageSize":  pageSize,
+				"pageIndex": i,
+			}).Logger.Errorf("Failed to get page of servers, attempting to continue: %s", err.Error())
 
-func (c *Client) GatherServersNonBlocking(pageSize, inFlightPages int) (chan *fleetdbapi.Server, *semaphore.Weighted, error) {
-	serverCh := make(chan *fleetdbapi.Server)
-	concLimiter := semaphore.NewWeighted(int64(inFlightPages * pageSize))
+			continue
+		}
 
-	go func() {
-		c.gatherServers(pageSize, serverCh, concLimiter)
-	}()
+		c.log.WithFields(logrus.Fields{
+			"index":      i,
+			"iterations": totalPages,
+			"got":        len(servers),
+		}).Trace("Got server page")
+
+		for i := range servers {
+			err = c.CreateConditionInventory(servers[i].UUID)
+			if err != nil {
+				return err
+			}
+		}
+	}
 
-	return serverCh, concLimiter, nil
+	return nil
 }