change(syncer): parallel via waitgroup

Author: l3uddz

cmd/sync.go

@@ -10,10 +10,12 @@ import (
 	"github.com/sirupsen/logrus"
 	"github.com/spf13/cobra"
 	"strings"
+	"sync"
 )
 
 var (
-	flagSyncer string
+	flagSyncer      string
+	flagParallelism int
 )
 
 var syncCmd = &cobra.Command{
@@ -27,7 +29,12 @@ var syncCmd = &cobra.Command{
 		defer cache.Close()
 
 		// iterate syncer's
+		var wg sync.WaitGroup
+		pos := 0
+
 		for _, syncerConfig := range config.Config.Syncer {
+			syncerConfig := syncerConfig
+
 			log := log.WithField("syncer", syncerConfig.Name)
 
 			// skip disabled syncer(s)
@@ -43,54 +50,71 @@ var syncCmd = &cobra.Command{
 			}
 
 			// create syncer
-			sync, err := syncer.New(config.Config, &syncerConfig, syncerConfig.Name)
+			syncr, err := syncer.New(config.Config, &syncerConfig, syncerConfig.Name)
 			if err != nil {
 				log.WithError(err).Error("Failed initializing syncer, skipping...")
 				continue
 			}
 
-			serviceAccountCount := sync.RemoteServiceAccountFiles.ServiceAccountsCount()
+			serviceAccountCount := syncr.RemoteServiceAccountFiles.ServiceAccountsCount()
 			if serviceAccountCount > 0 {
-				sync.Log.WithField("found_files", serviceAccountCount).Info("Loaded service accounts")
+				syncr.Log.WithField("found_files", serviceAccountCount).Info("Loaded service accounts")
 			} else {
 				// no service accounts were loaded
 				// check to see if any of the copy or sync remote(s) are banned
-				banned, expiry := rclone.AnyRemotesBanned(sync.Config.Remotes.Copy)
+				banned, expiry := rclone.AnyRemotesBanned(syncr.Config.Remotes.Copy)
 				if banned && !expiry.IsZero() {
 					// one of the copy remotes is banned, abort
-					sync.Log.WithFields(logrus.Fields{
+					syncr.Log.WithFields(logrus.Fields{
 						"expires_time": expiry,
 						"expires_in":   humanize.Time(expiry),
 					}).Warn("Cannot proceed with sync as a copy remote is banned")
 					continue
 				}
 
-				banned, expiry = rclone.AnyRemotesBanned(sync.Config.Remotes.Sync)
+				banned, expiry = rclone.AnyRemotesBanned(syncr.Config.Remotes.Sync)
 				if banned && !expiry.IsZero() {
 					// one of the sync remotes is banned, abort
-					sync.Log.WithFields(logrus.Fields{
+					syncr.Log.WithFields(logrus.Fields{
 						"expires_time": expiry,
 						"expires_in":   humanize.Time(expiry),
 					}).Warn("Cannot proceed with sync as a sync remote is banned")
 					continue
 				}
 			}
 
+			pos += 1
 			log.Info("Syncer commencing...")
 
 			// perform sync
-			if err := performSync(sync); err != nil {
-				sync.Log.WithError(err).Error("Error occurred while running syncer, skipping...")
-				continue
+			wg.Add(1)
+			go func(s *syncer.Syncer, w *sync.WaitGroup) {
+				defer w.Done()
+
+				// run syncer
+				if err := performSync(s); err != nil {
+					s.Log.WithError(err).Error("Error occurred while running syncer, skipping...")
+				}
+			}(syncr, &wg)
+
+			// parallelism limit reached?
+			if pos%flagParallelism == 0 {
+				log.Info("Waiting before starting next syncer")
+				wg.Wait()
 			}
 		}
+
+		// wait for all syncers to finish
+		log.Info("Waiting for syncer(s) to finish")
+		wg.Wait()
 	},
 }
 
 func init() {
 	rootCmd.AddCommand(syncCmd)
 
 	syncCmd.Flags().StringVarP(&flagSyncer, "syncer", "s", "", "Run for a specific syncer")
+	syncCmd.Flags().IntVarP(&flagParallelism, "parallelism", "p", 1, "Max parallel syncers")
 }
 
 func performSync(s *syncer.Syncer) error {

rclone/sa.go

@@ -13,6 +13,7 @@ import (
 	"sort"
 	"strconv"
 	"strings"
+	"sync"
 	"time"
 )
 
@@ -34,6 +35,17 @@ type ServiceAccountManager struct {
 	remoteServiceAccounts       map[string]RemoteServiceAccounts
 }
 
+var (
+	mtx  sync.Mutex
+	psac map[string]time.Time
+)
+
+/* Private */
+
+func init() {
+	psac = make(map[string]time.Time)
+}
+
 /* Public */
 
 func NewServiceAccountManager(serviceAccountFolders map[string]string) *ServiceAccountManager {
@@ -123,6 +135,10 @@ func (m *ServiceAccountManager) GetServiceAccount(remotePaths ...string) ([]*Rem
 	var err error
 	successfulRemotes := make(map[string]*types.Nil)
 
+	// acquire global lock
+	mtx.Lock()
+	defer mtx.Unlock()
+
 	for _, remotePath := range remotePaths {
 		saFound := false
 
@@ -155,6 +171,19 @@ func (m *ServiceAccountManager) GetServiceAccount(remotePaths ...string) ([]*Rem
 				continue
 			}
 
+			// has this service account been issued within N seconds?
+			expiry, exists := psac[sa.RealPath]
+			switch {
+			case exists && expiry.Before(time.Now().UTC()):
+				// it was issued before, but it was not within N seconds
+				delete(psac, sa.RealPath)
+			case exists:
+				// it was issued before and it has not expired yet
+				continue
+			default:
+				break
+			}
+
 			// this service account is unbanned
 			serviceAccounts = append(serviceAccounts, &RemoteServiceAccount{
 				RemoteEnvVar:       remote.RemoteEnvVar,
@@ -177,6 +206,13 @@ func (m *ServiceAccountManager) GetServiceAccount(remotePaths ...string) ([]*Rem
 		break
 	}
 
+	// were service accounts found?
+	if len(serviceAccounts) > 0 {
+		for _, sa := range serviceAccounts {
+			psac[sa.ServiceAccountPath] = time.Now().UTC().Add(10 * time.Second)
+		}
+	}
+
 	return serviceAccounts, err
 }