Add max_parallel to mssql and postgresql

physical/couchdb.go

@@ -8,10 +8,12 @@ import (
 	"net/http"
 	"net/url"
 	"os"
+	"strconv"
 	"strings"
 	"time"
 
 	"github.com/armon/go-metrics"
+	"github.com/hashicorp/errwrap"
 	cleanhttp "github.com/hashicorp/go-cleanhttp"
 	log "github.com/mgutz/logxi/v1"
 )
@@ -160,6 +162,19 @@ func buildCouchDBBackend(conf map[string]string, logger log.Logger) (*CouchDBBac
 		password = conf["password"]
 	}
 
+	maxParStr, ok := conf["max_parallel"]
+	var maxParInt int
+	var err error
+	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("couchdb: max_parallel set", "max_parallel", maxParInt)
+		}
+	}
+
 	return &CouchDBBackend{
 		client: &couchDBClient{
 			endpoint: endpoint,
@@ -168,7 +183,7 @@ func buildCouchDBBackend(conf map[string]string, logger log.Logger) (*CouchDBBac
 			Client:   cleanhttp.DefaultPooledClient(),
 		},
 		logger:     logger,
-		permitPool: NewPermitPool(DefaultParallelOperations),
+		permitPool: NewPermitPool(maxParInt),
 	}, nil
 }
 
@@ -211,6 +226,9 @@ func (m *CouchDBBackend) Delete(key string) error {
 func (m *CouchDBBackend) List(prefix string) ([]string, error) {
 	defer metrics.MeasureSince([]string{"couchdb", "list"}, time.Now())
 
+	m.permitPool.Acquire()
+	defer m.permitPool.Release()
+
 	items, err := m.client.list(prefix)
 	if err != nil {
 		return nil, err

physical/mssql.go

@@ -4,11 +4,13 @@ import (
 	"database/sql"
 	"fmt"
 	"sort"
+	"strconv"
 	"strings"
 	"time"
 
 	"github.com/armon/go-metrics"
 	_ "github.com/denisenkom/go-mssqldb"
+	"github.com/hashicorp/errwrap"
 	"github.com/hashicorp/vault/helper/strutil"
 	log "github.com/mgutz/logxi/v1"
 )
@@ -18,6 +20,7 @@ type MsSQLBackend struct {
 	client     *sql.DB
 	statements map[string]*sql.Stmt
 	logger     log.Logger
+	permitPool *PermitPool
 }
 
 func newMsSQLBackend(conf map[string]string, logger log.Logger) (Backend, error) {
@@ -36,6 +39,21 @@ func newMsSQLBackend(conf map[string]string, logger log.Logger) (Backend, error)
 		return nil, fmt.Errorf("missing server")
 	}
 
+	maxParStr, ok := conf["max_parallel"]
+	var maxParInt int
+	var err error
+	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("mysql: max_parallel set", "max_parallel", maxParInt)
+		}
+	} else {
+		maxParInt = DefaultParallelOperations
+	}
+
 	database, ok := conf["database"]
 	if !ok {
 		database = "Vault"
@@ -80,6 +98,8 @@ func newMsSQLBackend(conf map[string]string, logger log.Logger) (Backend, error)
 		return nil, fmt.Errorf("failed to connect to mssql: %v", err)
 	}
 
+	db.SetMaxOpenConns(maxParInt)
+
 	if _, err := db.Exec("IF NOT EXISTS(SELECT * FROM sys.databases WHERE name = '" + database + "') CREATE DATABASE " + database); err != nil {
 		return nil, fmt.Errorf("failed to create mssql database: %v", err)
 	}
@@ -116,6 +136,7 @@ func newMsSQLBackend(conf map[string]string, logger log.Logger) (Backend, error)
 		client:     db,
 		statements: make(map[string]*sql.Stmt),
 		logger:     logger,
+		permitPool: NewPermitPool(maxParInt),
 	}
 
 	statements := map[string]string{
@@ -149,6 +170,9 @@ func (m *MsSQLBackend) prepare(name, query string) error {
 func (m *MsSQLBackend) Put(entry *Entry) error {
 	defer metrics.MeasureSince([]string{"mssql", "put"}, time.Now())
 
+	m.permitPool.Acquire()
+	defer m.permitPool.Release()
+
 	_, err := m.statements["put"].Exec(entry.Key, entry.Value, entry.Key, entry.Key, entry.Value)
 	if err != nil {
 		return err
@@ -160,6 +184,9 @@ func (m *MsSQLBackend) Put(entry *Entry) error {
 func (m *MsSQLBackend) Get(key string) (*Entry, error) {
 	defer metrics.MeasureSince([]string{"mssql", "get"}, time.Now())
 
+	m.permitPool.Acquire()
+	defer m.permitPool.Release()
+
 	var result []byte
 	err := m.statements["get"].QueryRow(key).Scan(&result)
 	if err == sql.ErrNoRows {
@@ -181,6 +208,9 @@ func (m *MsSQLBackend) Get(key string) (*Entry, error) {
 func (m *MsSQLBackend) Delete(key string) error {
 	defer metrics.MeasureSince([]string{"mssql", "delete"}, time.Now())
 
+	m.permitPool.Acquire()
+	defer m.permitPool.Release()
+
 	_, err := m.statements["delete"].Exec(key)
 	if err != nil {
 		return err
@@ -192,6 +222,9 @@ func (m *MsSQLBackend) Delete(key string) error {
 func (m *MsSQLBackend) List(prefix string) ([]string, error) {
 	defer metrics.MeasureSince([]string{"mssql", "list"}, time.Now())
 
+	m.permitPool.Acquire()
+	defer m.permitPool.Release()
+
 	likePrefix := prefix + "%"
 	rows, err := m.statements["list"].Query(likePrefix)
 	if err != nil {

physical/mysql.go

@@ -76,6 +76,8 @@ func newMySQLBackend(conf map[string]string, logger log.Logger) (Backend, error)
 		if logger.IsDebug() {
 			logger.Debug("mysql: max_parallel set", "max_parallel", maxParInt)
 		}
+	} else {
+		maxParInt = DefaultParallelOperations
 	}
 
 	dsnParams := url.Values{}
@@ -95,6 +97,8 @@ func newMySQLBackend(conf map[string]string, logger log.Logger) (Backend, error)
 		return nil, fmt.Errorf("failed to connect to mysql: %v", err)
 	}
 
+	db.SetMaxOpenConns(maxParInt)
+
 	// Create the required database if it doesn't exists.
 	if _, err := db.Exec("CREATE DATABASE IF NOT EXISTS " + database); err != nil {
 		return nil, fmt.Errorf("failed to create mysql database: %v", err)

physical/postgresql.go

@@ -3,9 +3,11 @@ package physical
 import (
 	"database/sql"
 	"fmt"
+	"strconv"
 	"strings"
 	"time"
 
+	"github.com/hashicorp/errwrap"
 	log "github.com/mgutz/logxi/v1"
 
 	"github.com/armon/go-metrics"
@@ -22,6 +24,7 @@ type PostgreSQLBackend struct {
 	delete_query string
 	list_query   string
 	logger       log.Logger
+	permitPool   *PermitPool
 }
 
 // newPostgreSQLBackend constructs a PostgreSQL backend using the given
@@ -39,11 +42,27 @@ func newPostgreSQLBackend(conf map[string]string, logger log.Logger) (Backend, e
 	}
 	quoted_table := pq.QuoteIdentifier(unquoted_table)
 
+	maxParStr, ok := conf["max_parallel"]
+	var maxParInt int
+	var err error
+	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("postgres: max_parallel set", "max_parallel", maxParInt)
+		}
+	} else {
+		maxParInt = DefaultParallelOperations
+	}
+
 	// Create PostgreSQL handle for the database.
 	db, err := sql.Open("postgres", connURL)
 	if err != nil {
 		return nil, fmt.Errorf("failed to connect to postgres: %v", err)
 	}
+	db.SetMaxOpenConns(maxParInt)
 
 	// Determine if we should use an upsert function (versions < 9.5)
 	var upsert_required bool
@@ -73,7 +92,8 @@ func newPostgreSQLBackend(conf map[string]string, logger log.Logger) (Backend, e
 		list_query: "SELECT key FROM " + quoted_table + " WHERE path = $1" +
 			"UNION SELECT DISTINCT substring(substr(path, length($1)+1) from '^.*?/') FROM " +
 			quoted_table + " WHERE parent_path LIKE $1 || '%'",
-		logger: logger,
+		logger:     logger,
+		permitPool: NewPermitPool(maxParInt),
 	}
 
 	return m, nil
@@ -107,6 +127,9 @@ func (m *PostgreSQLBackend) splitKey(fullPath string) (string, string, string) {
 func (m *PostgreSQLBackend) Put(entry *Entry) error {
 	defer metrics.MeasureSince([]string{"postgres", "put"}, time.Now())
 
+	m.permitPool.Acquire()
+	defer m.permitPool.Release()
+
 	parentPath, path, key := m.splitKey(entry.Key)
 
 	_, err := m.client.Exec(m.put_query, parentPath, path, key, entry.Value)
@@ -120,6 +143,9 @@ func (m *PostgreSQLBackend) Put(entry *Entry) error {
 func (m *PostgreSQLBackend) Get(fullPath string) (*Entry, error) {
 	defer metrics.MeasureSince([]string{"postgres", "get"}, time.Now())
 
+	m.permitPool.Acquire()
+	defer m.permitPool.Release()
+
 	_, path, key := m.splitKey(fullPath)
 
 	var result []byte
@@ -142,6 +168,9 @@ func (m *PostgreSQLBackend) Get(fullPath string) (*Entry, error) {
 func (m *PostgreSQLBackend) Delete(fullPath string) error {
 	defer metrics.MeasureSince([]string{"postgres", "delete"}, time.Now())
 
+	m.permitPool.Acquire()
+	defer m.permitPool.Release()
+
 	_, path, key := m.splitKey(fullPath)
 
 	_, err := m.client.Exec(m.delete_query, path, key)
@@ -156,6 +185,9 @@ func (m *PostgreSQLBackend) Delete(fullPath string) error {
 func (m *PostgreSQLBackend) List(prefix string) ([]string, error) {
 	defer metrics.MeasureSince([]string{"postgres", "list"}, time.Now())
 
+	m.permitPool.Acquire()
+	defer m.permitPool.Release()
+
 	rows, err := m.client.Query(m.list_query, "/"+prefix)
 	if err != nil {
 		return nil, err

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

@@ -30,13 +30,16 @@ storage "couchdb" {
 
 ## `couchdb` Parameters
 
-- `endpoint` `(string: "")` – Specifies your CouchDB endpoint. This can also be provided via the
-  environment variable `COUCHDB_ENDPOINT`.
+- `endpoint` `(string: "")` – Specifies your CouchDB endpoint. This can also be
+  provided via the environment variable `COUCHDB_ENDPOINT`.
 
-- `username` `(string: "")` – Specifies the user to authenticate as. This can also be provided via the
-  environment variable `COUCHDB_USERNAME`.
+- `username` `(string: "")` – Specifies the user to authenticate as. This can
+  also be provided via the environment variable `COUCHDB_USERNAME`.
 
-- `password` `(string: "")` – Specifies the user to authenticate as. This can also be provided via the
-  environment variable `COUCHDB_PASSWORD`.
+- `password` `(string: "")` – Specifies the user to authenticate as. This can
+  also be provided via the environment variable `COUCHDB_PASSWORD`.
+
+- `max_parallel` `(string: "128")` – Specifies the maximum number of concurrent
+  requests to CouchDB.
 
 [couchdb]: http://couchdb.apache.org/

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

@@ -39,7 +39,7 @@ storage "mssql" {
 - `username` `(string: "")` - enter the SQL Server Authentication user id or 
   the Windows Authentication user id in the DOMAIN\User format. 
   On Windows, if user id is empty or missing Single-Sign-On is used.
-  
+
 - `password` `(string: "")` – specifies the MSSQL password to connect to
   the database.
 
@@ -48,7 +48,7 @@ storage "mssql" {
 
 - `table` `(string: "Vault")` – Specifies the name of the table. If the table
   does not exist, Vault will attempt to create it.
-  
+
 - `schema` `(string: "dbo")` – Specifies the name of the schema. If the schema
   does not exist, Vault will attempt to create it.
 
@@ -58,6 +58,9 @@ storage "mssql" {
 
 - `logLevel` `(int: 0)` – logging flags (default 0/no logging, 63 for full logging) .
 
+- `max_parallel` `(string: "128")` – Specifies the maximum number of concurrent
+  requests to MSSQL.
+
 ## `mssql` Examples
 
 ### Custom Database, Table and Schema

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

@@ -82,6 +82,9 @@ LANGUAGE plpgsql;
   which to write Vault data. This table must already exist (Vault will not
   attempt to create it).
 
+- `max_parallel` `(string: "128")` – Specifies the maximum number of concurrent
+  requests to PostgreSQL.
+
 ## `postgresql` Examples
 
 ### Custom SSL Verification