Added ssh connection rate limiting feature

- allows enabling ssh connection rate limiting
- adds configurable amount of max connections per duration window
- adds configurable duration window

bosh/build_client.go

@@ -2,6 +2,7 @@ package bosh
 
 import (
 	"github.com/cloudfoundry-incubator/bosh-backup-and-restore/instance"
+	"github.com/cloudfoundry-incubator/bosh-backup-and-restore/ratelimiter"
 	"github.com/cloudfoundry-incubator/bosh-backup-and-restore/ssh"
 	"github.com/cloudfoundry/bosh-cli/v7/director"
 	"github.com/pkg/errors"
@@ -10,7 +11,7 @@ import (
 	boshlog "github.com/cloudfoundry/bosh-utils/logger"
 )
 
-func BuildClient(targetUrl, username, password, caCert, bbrVersion string, logger boshlog.Logger) (Client, error) {
+func BuildClient(targetUrl, username, password, caCert, bbrVersion string, rateLimiter ratelimiter.RateLimiter, logger boshlog.Logger) (Client, error) {
 	var client Client
 
 	factoryConfig, err := director.NewConfigFromURL(targetUrl)
@@ -44,7 +45,7 @@ func BuildClient(targetUrl, username, password, caCert, bbrVersion string, logge
 		return client, errors.Wrap(err, "error building bosh director client")
 	}
 
-	return NewClient(boshDirector, director.NewSSHOpts, ssh.NewSshRemoteRunner, logger, instance.NewJobFinder(bbrVersion, logger), NewBoshManifestQuerier), nil
+	return NewClient(boshDirector, director.NewSSHOpts, ssh.NewSshRemoteRunner, rateLimiter, logger, instance.NewJobFinder(bbrVersion, logger), NewBoshManifestQuerier), nil
 }
 
 func getDirectorInfo(directorFactory director.Factory, factoryConfig director.FactoryConfig) (director.Info, error) {

bosh/build_client_test.go

@@ -8,6 +8,7 @@ import (
 	"github.com/cloudfoundry-incubator/bosh-backup-and-restore/internal/cf-webmock/mockbosh"
 	"github.com/cloudfoundry-incubator/bosh-backup-and-restore/internal/cf-webmock/mockhttp"
 	"github.com/cloudfoundry-incubator/bosh-backup-and-restore/internal/cf-webmock/mockuaa"
+	"github.com/cloudfoundry-incubator/bosh-backup-and-restore/ratelimiter"
 	boshlog "github.com/cloudfoundry/bosh-utils/logger"
 	. "github.com/onsi/ginkgo/v2"
 	. "github.com/onsi/gomega"
@@ -50,7 +51,7 @@ var _ = Describe("BuildClient", func() {
 				mockbosh.Manifest(deploymentName).RespondsWith([]byte("manifest contents")),
 			)
 
-			client, err := BuildClient(director.URL, username, password, caCert, bbrVersion, logger)
+			client, err := BuildClient(director.URL, username, password, caCert, bbrVersion, ratelimiter.NoOpRateLimiter{}, logger)
 
 			Expect(err).NotTo(HaveOccurred())
 			manifest, err := client.GetManifest(deploymentName)
@@ -75,7 +76,7 @@ var _ = Describe("BuildClient", func() {
 				mockbosh.Manifest(deploymentName).RespondsWith([]byte("manifest contents")),
 			)
 
-			client, err := BuildClient(director.URL, username, password, caCert, bbrVersion, logger)
+			client, err := BuildClient(director.URL, username, password, caCert, bbrVersion, ratelimiter.NoOpRateLimiter{}, logger)
 
 			Expect(err).NotTo(HaveOccurred())
 			manifest, err := client.GetManifest(deploymentName)
@@ -90,7 +91,7 @@ var _ = Describe("BuildClient", func() {
 			director.VerifyAndMock(
 				mockbosh.Info().WithAuthTypeUAA(""),
 			)
-			_, err := BuildClient(director.URL, username, password, caCert, bbrVersion, logger)
+			_, err := BuildClient(director.URL, username, password, caCert, bbrVersion, ratelimiter.NoOpRateLimiter{}, logger)
 
 			Expect(err).To(MatchError(ContainSubstring("invalid UAA URL")))
 
@@ -103,7 +104,7 @@ var _ = Describe("BuildClient", func() {
 		caCertPath := "-----BEGIN"
 		basicAuthDirectorURL := director.URL
 
-		_, err := BuildClient(basicAuthDirectorURL, username, password, caCertPath, bbrVersion, logger)
+		_, err := BuildClient(basicAuthDirectorURL, username, password, caCertPath, bbrVersion, ratelimiter.NoOpRateLimiter{}, logger)
 		Expect(err).To(MatchError(ContainSubstring("Missing PEM block")))
 	})
 
@@ -113,7 +114,7 @@ var _ = Describe("BuildClient", func() {
 		caCertPath := ""
 		basicAuthDirectorURL := ""
 
-		_, err := BuildClient(basicAuthDirectorURL, username, password, caCertPath, bbrVersion, logger)
+		_, err := BuildClient(basicAuthDirectorURL, username, password, caCertPath, bbrVersion, ratelimiter.NoOpRateLimiter{}, logger)
 		Expect(err).To(MatchError(ContainSubstring("invalid bosh URL")))
 	})
 
@@ -125,7 +126,7 @@ var _ = Describe("BuildClient", func() {
 			mockbosh.Info().Fails("fooo!"),
 		)
 
-		_, err := BuildClient(director.URL, username, password, caCert, bbrVersion, logger)
+		_, err := BuildClient(director.URL, username, password, caCert, bbrVersion, ratelimiter.NoOpRateLimiter{}, logger)
 		Expect(err).To(MatchError(ContainSubstring("bosh director unreachable or unhealthy")))
 	})
 })

bosh/client.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/cloudfoundry-incubator/bosh-backup-and-restore/instance"
 	"github.com/cloudfoundry-incubator/bosh-backup-and-restore/orchestrator"
+	"github.com/cloudfoundry-incubator/bosh-backup-and-restore/ratelimiter"
 	"github.com/cloudfoundry-incubator/bosh-backup-and-restore/ssh"
 	"github.com/cloudfoundry/bosh-cli/v7/director"
 	"github.com/cloudfoundry/bosh-utils/uuid"
@@ -25,13 +26,15 @@ type BoshClient interface {
 func NewClient(boshDirector director.Director,
 	sshOptsGenerator ssh.SSHOptsGenerator,
 	remoteRunnerFactory ssh.RemoteRunnerFactory,
+	rateLimiter ratelimiter.RateLimiter,
 	logger Logger,
 	jobFinder instance.JobFinder,
 	manifestQuerierCreator instance.ManifestQuerierCreator) Client {
 	return Client{
 		Director:               boshDirector,
 		SSHOptsGenerator:       sshOptsGenerator,
 		RemoteRunnerFactory:    remoteRunnerFactory,
+		RateLimiter:            rateLimiter,
 		Logger:                 logger,
 		jobFinder:              jobFinder,
 		manifestQuerierCreator: manifestQuerierCreator,
@@ -42,6 +45,7 @@ type Client struct {
 	director.Director
 	ssh.SSHOptsGenerator
 	ssh.RemoteRunnerFactory
+	ratelimiter.RateLimiter
 	Logger
 	jobFinder              instance.JobFinder
 	manifestQuerierCreator instance.ManifestQuerierCreator
@@ -112,7 +116,7 @@ func (c Client) FindInstances(deploymentName string) ([]orchestrator.Instance, e
 				return nil, errors.Wrap(err, "ssh.NewConnection.ParseAuthorizedKey failed")
 			}
 
-			remoteRunner, err := c.RemoteRunnerFactory(host.Host, host.Username, privateKey, gossh.FixedHostKey(hostPublicKey), supportedEncryptionAlgorithms(hostPublicKey), c.Logger)
+			remoteRunner, err := c.RemoteRunnerFactory(host.Host, host.Username, privateKey, gossh.FixedHostKey(hostPublicKey), supportedEncryptionAlgorithms(hostPublicKey), c.RateLimiter, c.Logger)
 			if err != nil {
 				cleanupAlreadyMadeConnections(deployment, slugs, sshOpts)
 				return nil, errors.Wrap(err, "failed to connect using ssh")

bosh/client_test.go

@@ -12,6 +12,7 @@ import (
 	"github.com/cloudfoundry-incubator/bosh-backup-and-restore/instance"
 	instancefakes "github.com/cloudfoundry-incubator/bosh-backup-and-restore/instance/fakes"
 	"github.com/cloudfoundry-incubator/bosh-backup-and-restore/orchestrator"
+	"github.com/cloudfoundry-incubator/bosh-backup-and-restore/ratelimiter"
 	"github.com/cloudfoundry-incubator/bosh-backup-and-restore/ssh"
 	sshfakes "github.com/cloudfoundry-incubator/bosh-backup-and-restore/ssh/fakes"
 	"github.com/cloudfoundry/bosh-cli/v7/director"
@@ -43,7 +44,7 @@ var _ = Describe("Director", func() {
 	var b bosh.BoshClient
 
 	JustBeforeEach(func() {
-		b = bosh.NewClient(boshDirector, optsGenerator.Spy, remoteRunnerFactory.Spy, boshLogger, fakeJobFinder, manifestQuerierCreator.Spy)
+		b = bosh.NewClient(boshDirector, optsGenerator.Spy, remoteRunnerFactory.Spy, ratelimiter.NoOpRateLimiter{}, boshLogger, fakeJobFinder, manifestQuerierCreator.Spy)
 	})
 
 	BeforeEach(func() {
@@ -166,7 +167,7 @@ var _ = Describe("Director", func() {
 
 			It("creates a remote runner for each host", func() {
 				Expect(remoteRunnerFactory.CallCount()).To(Equal(1))
-				host, username, privateKey, _, hostPublicKeyAlgorithm, logger := remoteRunnerFactory.ArgsForCall(0)
+				host, username, privateKey, _, hostPublicKeyAlgorithm, _, logger := remoteRunnerFactory.ArgsForCall(0)
 				Expect(host).To(Equal("10.0.0.0"))
 				Expect(username).To(Equal("username"))
 				Expect(privateKey).To(Equal("private_key"))
@@ -195,7 +196,7 @@ var _ = Describe("Director", func() {
 
 			It("uses the specified port", func() {
 				Expect(remoteRunnerFactory.CallCount()).To(Equal(1))
-				host, username, privateKey, _, hostPublicKeyAlgorithm, logger := remoteRunnerFactory.ArgsForCall(0)
+				host, username, privateKey, _, hostPublicKeyAlgorithm, _, logger := remoteRunnerFactory.ArgsForCall(0)
 				Expect(host).To(Equal("10.0.0.0:3457"))
 				Expect(username).To(Equal("username"))
 				Expect(privateKey).To(Equal("private_key"))
@@ -328,14 +329,14 @@ var _ = Describe("Director", func() {
 			It("creates a remote runner for each host", func() {
 				Expect(remoteRunnerFactory.CallCount()).To(Equal(2))
 
-				host, username, privateKey, _, hostPublicKeyAlgorithm, logger := remoteRunnerFactory.ArgsForCall(0)
+				host, username, privateKey, _, hostPublicKeyAlgorithm, _, logger := remoteRunnerFactory.ArgsForCall(0)
 				Expect(host).To(Equal("10.0.0.1"))
 				Expect(username).To(Equal("username"))
 				Expect(privateKey).To(Equal("private_key"))
 				Expect(hostPublicKeyAlgorithm).To(Equal(hostKeyAlgorithmRSA))
 				Expect(logger).To(Equal(boshLogger))
 
-				host, username, privateKey, _, hostPublicKeyAlgorithm, logger = remoteRunnerFactory.ArgsForCall(1)
+				host, username, privateKey, _, hostPublicKeyAlgorithm, _, logger = remoteRunnerFactory.ArgsForCall(1)
 				Expect(host).To(Equal("10.0.0.2"))
 				Expect(username).To(Equal("username"))
 				Expect(privateKey).To(Equal("private_key"))
@@ -621,21 +622,21 @@ var _ = Describe("Director", func() {
 			It("creates a remote runner for each host that has scripts, and the first instance of each group that doesn't", func() {
 				Expect(remoteRunnerFactory.CallCount()).To(Equal(3))
 
-				host, username, privateKey, _, hostPublicKeyAlgorithm, logger := remoteRunnerFactory.ArgsForCall(0)
+				host, username, privateKey, _, hostPublicKeyAlgorithm, _, logger := remoteRunnerFactory.ArgsForCall(0)
 				Expect(host).To(Equal("10.0.0.1"))
 				Expect(username).To(Equal("username"))
 				Expect(privateKey).To(Equal("private_key"))
 				Expect(hostPublicKeyAlgorithm).To(Equal(hostKeyAlgorithmRSA))
 				Expect(logger).To(Equal(boshLogger))
 
-				host, username, privateKey, _, hostPublicKeyAlgorithm, logger = remoteRunnerFactory.ArgsForCall(1)
+				host, username, privateKey, _, hostPublicKeyAlgorithm, _, logger = remoteRunnerFactory.ArgsForCall(1)
 				Expect(host).To(Equal("10.0.0.3"))
 				Expect(username).To(Equal("username"))
 				Expect(privateKey).To(Equal("private_key"))
 				Expect(hostPublicKeyAlgorithm).To(Equal(hostKeyAlgorithmRSA))
 				Expect(logger).To(Equal(boshLogger))
 
-				host, username, privateKey, _, hostPublicKeyAlgorithm, logger = remoteRunnerFactory.ArgsForCall(2)
+				host, username, privateKey, _, hostPublicKeyAlgorithm, _, logger = remoteRunnerFactory.ArgsForCall(2)
 				Expect(host).To(Equal("10.0.0.4"))
 				Expect(username).To(Equal("username"))
 				Expect(privateKey).To(Equal("private_key"))
@@ -689,7 +690,7 @@ var _ = Describe("Director", func() {
 
 			It("uses the ECDSA algorithm to create its remote runners", func() {
 				Expect(remoteRunnerFactory.CallCount()).To(Equal(1))
-				_, _, _, _, hostPublicKeyAlgorithm, _ := remoteRunnerFactory.ArgsForCall(0)
+				_, _, _, _, hostPublicKeyAlgorithm, _, _ := remoteRunnerFactory.ArgsForCall(0)
 				Expect(hostPublicKeyAlgorithm).To(Equal(hostKeyAlgorithmECDSA))
 			})
 
@@ -995,7 +996,7 @@ var _ = Describe("Director", func() {
 						}}, nil
 					}
 
-					remoteRunnerFactory.Stub = func(host, user, privateKey string, publicKeyCallback gossh.HostKeyCallback, publicKeyAlgorithm []string, logger ssh.Logger) (ssh.RemoteRunner, error) {
+					remoteRunnerFactory.Stub = func(host, user, privateKey string, publicKeyCallback gossh.HostKeyCallback, publicKeyAlgorithm []string, rateLimiter ratelimiter.RateLimiter, logger ssh.Logger) (ssh.RemoteRunner, error) {
 						if host == "10.0.0.0_job1" {
 							return remoteRunner, nil
 						}

cli/command/all_deployments.go

@@ -15,6 +15,7 @@ import (
 	"github.com/cloudfoundry-incubator/bosh-backup-and-restore/executor/deployment"
 	"github.com/cloudfoundry-incubator/bosh-backup-and-restore/factory"
 	"github.com/cloudfoundry-incubator/bosh-backup-and-restore/orchestrator"
+	"github.com/cloudfoundry-incubator/bosh-backup-and-restore/ratelimiter"
 	"github.com/urfave/cli"
 )
 
@@ -128,6 +129,18 @@ func getDeploymentParams(c *cli.Context) (string, string, string, string, string
 	return username, password, target, caCert, bbrVersion, debug, deployment, allDeployments
 }
 
+func getConnectionRateLimiter(c *cli.Context) (ratelimiter.RateLimiter, error) {
+	enabled := c.Parent().Bool("rate-limiting")
+	maxConnections := c.Parent().Int("rate-limiting-max-connections")
+	duration := c.Parent().String("rate-limiting-duration")
+
+	if enabled {
+		return ratelimiter.NewConnectionRateLimiter(maxConnections, duration)
+	}
+	return ratelimiter.NewNoOpRateLimiter(), nil
+
+}
+
 type DeploymentExecutable struct {
 	action ActionFunc
 	name   string

cli/command/deployment_backup.go

@@ -7,6 +7,7 @@ import (
 	"github.com/cloudfoundry-incubator/bosh-backup-and-restore/executor/deployment"
 	"github.com/cloudfoundry-incubator/bosh-backup-and-restore/factory"
 	"github.com/cloudfoundry-incubator/bosh-backup-and-restore/orchestrator"
+	"github.com/cloudfoundry-incubator/bosh-backup-and-restore/ratelimiter"
 	"github.com/urfave/cli"
 )
 
@@ -50,17 +51,23 @@ func (d DeploymentBackupCommand) Action(c *cli.Context) error {
 	unsafeLockFree := c.Bool("unsafe-lock-free")
 	artifactPath := c.String("artifact-path")
 
+	rateLimiter, err := getConnectionRateLimiter(c)
+
+	if err != nil {
+		return err
+	}
+
 	if allDeployments {
 		if unsafeLockFree {
 			return processError(orchestrator.NewError(fmt.Errorf("Cannot use the --unsafe-lock-free flag in conjunction with the --all-deployments flag")))
 		}
-		return backupAll(target, username, password, caCert, artifactPath, withManifest, bbrVersion, debug)
+		return backupAll(target, username, password, caCert, artifactPath, withManifest, bbrVersion, debug, rateLimiter)
 	}
 
-	return backupSingleDeployment(deployment, target, username, password, caCert, artifactPath, withManifest, bbrVersion, unsafeLockFree, debug)
+	return backupSingleDeployment(deployment, target, username, password, caCert, artifactPath, withManifest, bbrVersion, unsafeLockFree, debug, rateLimiter)
 }
 
-func backupAll(target, username, password, caCert, artifactPath string, withManifest bool, bbrVersion string, debug bool) error {
+func backupAll(target, username, password, caCert, artifactPath string, withManifest bool, bbrVersion string, debug bool, rateLimiter ratelimiter.RateLimiter) error {
 	backupAction := func(deploymentName string) orchestrator.Error {
 		timestamp := time.Now().UTC().Format(artifactTimeStampFormat)
 		logFilePath, buffer, logger, logErr := createLogger(timestamp, artifactPath, deploymentName, debug)
@@ -76,6 +83,7 @@ func backupAll(target, username, password, caCert, artifactPath string, withMani
 			withManifest,
 			false,
 			bbrVersion,
+			rateLimiter,
 			logger,
 			timestamp,
 		)
@@ -106,7 +114,7 @@ func backupAll(target, username, password, caCert, artifactPath string, withMani
 	fmt.Println("Starting backup...")
 
 	logger, _ := factory.BuildBoshLoggerWithCustomBuffer(debug)
-	boshClient, err := factory.BuildBoshClient(target, username, password, caCert, bbrVersion, logger)
+	boshClient, err := factory.BuildBoshClient(target, username, password, caCert, bbrVersion, rateLimiter, logger)
 	if err != nil {
 		return processError(orchestrator.NewError(err))
 	}
@@ -119,11 +127,11 @@ func backupAll(target, username, password, caCert, artifactPath string, withMani
 		deployment.NewParallelExecutor())
 }
 
-func backupSingleDeployment(deployment, target, username, password, caCert, artifactPath string, withManifest bool, bbrVersion string, unsafeLockFree, debug bool) error {
+func backupSingleDeployment(deployment, target, username, password, caCert, artifactPath string, withManifest bool, bbrVersion string, unsafeLockFree, debug bool, rateLimiter ratelimiter.RateLimiter) error {
 	logger := factory.BuildBoshLogger(debug)
 	timeStamp := time.Now().UTC().Format(artifactTimeStampFormat)
 
-	backuper, err := factory.BuildDeploymentBackuper(target, username, password, caCert, withManifest, unsafeLockFree, bbrVersion, logger, timeStamp)
+	backuper, err := factory.BuildDeploymentBackuper(target, username, password, caCert, withManifest, unsafeLockFree, bbrVersion, rateLimiter, logger, timeStamp)
 	if err != nil {
 		return processError(orchestrator.NewError(err))
 	}

cli/command/deployment_backup_cleanup.go

@@ -5,6 +5,7 @@ import (
 	"time"
 
 	"github.com/cloudfoundry-incubator/bosh-backup-and-restore/executor/deployment"
+	"github.com/cloudfoundry-incubator/bosh-backup-and-restore/ratelimiter"
 
 	"github.com/cloudfoundry-incubator/bosh-backup-and-restore/factory"
 	"github.com/cloudfoundry-incubator/bosh-backup-and-restore/orchestrator"
@@ -31,6 +32,12 @@ func (d DeploymentBackupCleanupCommand) Action(c *cli.Context) error {
 
 	username, password, target, caCert, bbrVersion, debug, deployment, allDeployments := getDeploymentParams(c)
 
+	rateLimiter, err := getConnectionRateLimiter(c)
+
+	if err != nil {
+		return err
+	}
+
 	if !allDeployments {
 		logger := factory.BuildBoshLogger(debug)
 
@@ -40,6 +47,7 @@ func (d DeploymentBackupCleanupCommand) Action(c *cli.Context) error {
 			password,
 			caCert,
 			c.App.Version,
+			rateLimiter,
 			logger,
 		)
 		if err != nil {
@@ -50,10 +58,10 @@ func (d DeploymentBackupCleanupCommand) Action(c *cli.Context) error {
 		return processError(cleanupErr)
 	}
 
-	return cleanupAllDeployments(target, username, password, caCert, bbrVersion, debug)
+	return cleanupAllDeployments(target, username, password, caCert, bbrVersion, debug, rateLimiter)
 }
 
-func cleanupAllDeployments(target, username, password, caCert, bbrVersion string, debug bool) error {
+func cleanupAllDeployments(target, username, password, caCert, bbrVersion string, debug bool, rateLimiter ratelimiter.RateLimiter) error {
 	cleanupAction := func(deploymentName string) orchestrator.Error {
 		timestamp := time.Now().UTC().Format(artifactTimeStampFormat)
 		logFilePath, buffer, logger, logErr := createLogger(timestamp, "", deploymentName, debug)
@@ -67,6 +75,7 @@ func cleanupAllDeployments(target, username, password, caCert, bbrVersion string
 			password,
 			caCert,
 			bbrVersion,
+			rateLimiter,
 			logger,
 		)
 
@@ -93,7 +102,7 @@ func cleanupAllDeployments(target, username, password, caCert, bbrVersion string
 
 	logger, _ := factory.BuildBoshLoggerWithCustomBuffer(debug)
 
-	boshClient, err := factory.BuildBoshClient(target, username, password, caCert, bbrVersion, logger)
+	boshClient, err := factory.BuildBoshClient(target, username, password, caCert, bbrVersion, rateLimiter, logger)
 	if err != nil {
 		return err
 	}

cli/command/deployment_pre_backup_check.go

@@ -30,13 +30,20 @@ func (d DeploymentPreBackupCheck) Cli() cli.Command {
 
 func (d DeploymentPreBackupCheck) Action(c *cli.Context) error {
 	username, password, target, caCert, bbrVersion, debug, deployment, allDeployments := getDeploymentParams(c)
+
+	rateLimiter, err := getConnectionRateLimiter(c)
+
+	if err != nil {
+		return err
+	}
+
 	var logger logger.Logger
 	if allDeployments {
 		logger, _ = factory.BuildBoshLoggerWithCustomBuffer(debug)
 	} else {
 		logger = factory.BuildBoshLogger(debug)
 	}
-	boshClient, err := factory.BuildBoshClient(target, username, password, caCert, bbrVersion, logger)
+	boshClient, err := factory.BuildBoshClient(target, username, password, caCert, bbrVersion, rateLimiter, logger)
 	if err != nil {
 		return processError(orchestrator.NewError(err))
 	}