Add permitPool support to S3

physical/s3.go

@@ -6,6 +6,7 @@ import (
 	"io"
 	"os"
 	"sort"
+	"strconv"
 	"strings"
 	"time"
 
@@ -16,6 +17,7 @@ import (
 	"github.com/aws/aws-sdk-go/aws/awserr"
 	"github.com/aws/aws-sdk-go/aws/session"
 	"github.com/aws/aws-sdk-go/service/s3"
+	"github.com/hashicorp/errwrap"
 	"github.com/hashicorp/vault/helper/awsutil"
 )
 
@@ -25,6 +27,7 @@ type S3Backend struct {
 	bucket string
 	client *s3.S3
 	logger log.Logger
+	permitPool *PermitPool
 }
 
 // newS3Backend constructs a S3 backend using a pre-existing
@@ -85,10 +88,23 @@ func newS3Backend(conf map[string]string, logger log.Logger) (Backend, error) {
 		return nil, fmt.Errorf("unable to access bucket '%s': %v", bucket, err)
 	}
 
+	maxParStr, ok := conf["max_parallel"]
+	var maxParInt int
+	if ok {
+		maxParInt, err = strconv.Atoi(maxParStr)
+		if err != nil {
+			return nil, errwrap.Wrapf("failed parsing max_parallel parameter: {{err}}", err)
+		}
+		if logger.IsDebug() {
+			logger.Debug("s3: max_parallel set", "max_parallel", maxParInt)
+		}
+	}
+
 	s := &S3Backend{
 		client: s3conn,
 		bucket: bucket,
 		logger: logger,
+		permitPool: NewPermitPool(maxParInt),
 	}
 	return s, nil
 }
@@ -97,6 +113,9 @@ func newS3Backend(conf map[string]string, logger log.Logger) (Backend, error) {
 func (s *S3Backend) Put(entry *Entry) error {
 	defer metrics.MeasureSince([]string{"s3", "put"}, time.Now())
 
+	s.permitPool.Acquire()
+	defer s.permitPool.Release()
+
 	_, err := s.client.PutObject(&s3.PutObjectInput{
 		Bucket: aws.String(s.bucket),
 		Key:    aws.String(entry.Key),
@@ -114,6 +133,9 @@ func (s *S3Backend) Put(entry *Entry) error {
 func (s *S3Backend) Get(key string) (*Entry, error) {
 	defer metrics.MeasureSince([]string{"s3", "get"}, time.Now())
 
+	s.permitPool.Acquire()
+	defer s.permitPool.Release()
+
 	resp, err := s.client.GetObject(&s3.GetObjectInput{
 		Bucket: aws.String(s.bucket),
 		Key:    aws.String(key),
@@ -122,9 +144,8 @@ func (s *S3Backend) Get(key string) (*Entry, error) {
 		// Return nil on 404s, error on anything else
 		if awsErr.StatusCode() == 404 {
 			return nil, nil
-		} else {
-			return nil, err
 		}
+		return nil, err
 	}
 	if err != nil {
 		return nil, err
@@ -151,6 +172,9 @@ func (s *S3Backend) Get(key string) (*Entry, error) {
 func (s *S3Backend) Delete(key string) error {
 	defer metrics.MeasureSince([]string{"s3", "delete"}, time.Now())
 
+	s.permitPool.Acquire()
+	defer s.permitPool.Release()
+
 	_, err := s.client.DeleteObject(&s3.DeleteObjectInput{
 		Bucket: aws.String(s.bucket),
 		Key:    aws.String(key),
@@ -168,6 +192,9 @@ func (s *S3Backend) Delete(key string) error {
 func (s *S3Backend) List(prefix string) ([]string, error) {
 	defer metrics.MeasureSince([]string{"s3", "list"}, time.Now())
 
+	s.permitPool.Acquire()
+	defer s.permitPool.Release()
+
 	params := &s3.ListObjectsV2Input{
 		Bucket: aws.String(s.bucket),
 		Prefix: aws.String(prefix),

website/source/docs/configuration/storage/s3.html.md

@@ -55,6 +55,9 @@ cause Vault to attempt to retrieve credentials from the AWS metadata service.
 - `session_token` `(string: "")` – Specifies the AWS session token. This can
   also be provided via the environment variable `AWS_SESSION_TOKEN`.
 
+- `max_parallel` `(int: 128)` – Specifies The maximum number of concurrent
+  requests to S3.
+
 ## `s3` Examples
 
 ### Default Example