Provider-SQL reduce statement and connection overhead

.gitmodules

@@ -1,3 +1,3 @@
 [submodule "build"]
 	path = build
-	url = https://github.com/upbound/build
+	url = https://github.com/crossplane/build

Makefile

@@ -124,7 +124,7 @@ dev: $(KIND) $(KUBECTL)
 	@$(KIND) create cluster --name=$(PROJECT_NAME)-dev
 	@$(KUBECTL) cluster-info --context kind-$(PROJECT_NAME)-dev
 	@$(INFO) Installing Crossplane CRDs
-	@$(KUBECTL) apply -k https://github.com/crossplane/crossplane//cluster?ref=master
+	@$(KUBECTL) apply -k https://github.com/crossplane/crossplane//cluster?ref=release-1.16 --server-side
 	@$(INFO) Installing Provider SQL CRDs
 	@$(KUBECTL) apply -R -f package/crds
 	@$(INFO) Starting Provider SQL controllers

apis/postgresql/v1alpha1/role_types.go

@@ -100,6 +100,8 @@ type RoleConfigurationParameter struct {
 
 // A RoleObservation represents the observed state of a PostgreSQL role.
 type RoleObservation struct {
+	// ConnectionLimit represents the applied connectionlimit
+	ConnectionLimit *int32 `json:"connectionLimit,omitempty"`
 	// PrivilegesAsClauses represents the applied privileges state, taking into account
 	// any defaults applied by Postgres, and expressed as a list of ROLE PRIVILEGE clauses.
 	PrivilegesAsClauses []string `json:"privilegesAsClauses,omitempty"`

package/crds/postgresql.sql.crossplane.io_roles.yaml

@@ -334,6 +334,10 @@ spec:
                           type: string
                       type: object
                     type: array
+                  connectionLimit:
+                    description: ConnectionLimit represents the applied connectionlimit
+                    format: int32
+                    type: integer
                   privilegesAsClauses:
                     description: |-
                       PrivilegesAsClauses represents the applied privileges state, taking into account

pkg/clients/postgresql/postgresql.go

@@ -5,6 +5,7 @@ import (
 	"database/sql"
 	"errors"
 	"net/url"
+	"time"
 
 	"github.com/crossplane-contrib/provider-sql/pkg/clients/xsql"
 	"github.com/lib/pq"
@@ -20,6 +21,8 @@ const (
 )
 
 type postgresDB struct {
+	db       *sql.DB
+	err      error
 	dsn      string
 	endpoint string
 	port     string
@@ -38,14 +41,46 @@ func New(creds map[string][]byte, database, sslmode string) xsql.DB {
 	password := string(creds[xpv1.ResourceCredentialsSecretPasswordKey])
 	dsn := DSN(username, password, endpoint, port, database, sslmode)
 
+	db, err := openDB(dsn, true)
+
 	return postgresDB{
+		db:       db,
+		err:      err,
 		dsn:      dsn,
 		endpoint: endpoint,
 		port:     port,
 		sslmode:  sslmode,
 	}
 }
 
+// openDB returns a new database connection
+func openDB(dsn string, setLimits bool) (*sql.DB, error) {
+	db, err := sql.Open("postgres", dsn)
+	if err != nil {
+		return nil, err
+	}
+
+	// Since we are now using connection pooling, establish some sensible defaults for connections
+	// Ideally these parameters would be set in the config section for the provider, but that
+	// can be deferred to a later time.
+	if setLimits {
+		db.SetMaxOpenConns(5)
+		db.SetMaxIdleConns(2)
+		db.SetConnMaxIdleTime(2 * time.Minute)
+		db.SetConnMaxLifetime(10 * time.Minute)
+	}
+
+	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	err = db.PingContext(ctx)
+	if err != nil {
+		return nil, err
+	}
+
+	return db, nil
+}
+
 // DSN returns the DSN URL
 func DSN(username, password, endpoint, port, database, sslmode string) string {
 	// Use net/url UserPassword to encode the username and password
@@ -63,20 +98,23 @@ func DSN(username, password, endpoint, port, database, sslmode string) string {
 // ExecTx executes an array of queries, committing if all are successful and
 // rolling back immediately on failure.
 func (c postgresDB) ExecTx(ctx context.Context, ql []xsql.Query) error {
-	d, err := sql.Open("postgres", c.dsn)
+	if c.db == nil || c.err != nil {
+		return c.err
+	}
+
+	err := c.db.PingContext(ctx)
 	if err != nil {
 		return err
 	}
 
-	tx, err := d.BeginTx(ctx, nil)
+	tx, err := c.db.BeginTx(ctx, nil)
 	if err != nil {
 		return err
 	}
 
 	// Rollback or Commit based on error state. Defer close in defer to make
 	// sure the connection is always closed.
 	defer func() {
-		defer d.Close() //nolint:errcheck
 		if err != nil {
 			tx.Rollback() //nolint:errcheck
 			return
@@ -94,37 +132,46 @@ func (c postgresDB) ExecTx(ctx context.Context, ql []xsql.Query) error {
 
 // Exec the supplied query.
 func (c postgresDB) Exec(ctx context.Context, q xsql.Query) error {
-	d, err := sql.Open("postgres", c.dsn)
+	if c.db == nil || c.err != nil {
+		return c.err
+	}
+
+	err := c.db.PingContext(ctx)
 	if err != nil {
 		return err
 	}
-	defer d.Close() //nolint:errcheck
 
-	_, err = d.ExecContext(ctx, q.String, q.Parameters...)
+	_, err = c.db.ExecContext(ctx, q.String, q.Parameters...)
 	return err
 }
 
 // Query the supplied query.
 func (c postgresDB) Query(ctx context.Context, q xsql.Query) (*sql.Rows, error) {
-	d, err := sql.Open("postgres", c.dsn)
+	if c.err != nil || c.db == nil {
+		return nil, c.err
+	}
+
+	err := c.db.PingContext(ctx)
 	if err != nil {
 		return nil, err
 	}
-	defer d.Close() //nolint:errcheck
 
-	rows, err := d.QueryContext(ctx, q.String, q.Parameters...)
+	rows, err := c.db.QueryContext(ctx, q.String, q.Parameters...)
 	return rows, err
 }
 
 // Scan the results of the supplied query into the supplied destination.
 func (c postgresDB) Scan(ctx context.Context, q xsql.Query, dest ...interface{}) error {
-	db, err := sql.Open("postgres", c.dsn)
+	if c.db == nil || c.err != nil {
+		return c.err
+	}
+
+	err := c.db.PingContext(ctx)
 	if err != nil {
 		return err
 	}
-	defer db.Close() //nolint:errcheck
 
-	return db.QueryRowContext(ctx, q.String, q.Parameters...).Scan(dest...)
+	return c.db.QueryRowContext(ctx, q.String, q.Parameters...).Scan(dest...)
 }
 
 // GetConnectionDetails returns the connection details for a user of this DB

pkg/controller/postgresql/role/reconciler.go

@@ -200,6 +200,7 @@ func (c *external) Observe(ctx context.Context, mg resource.Managed) (managed.Ex
 			Replication: new(bool),
 			BypassRls:   new(bool),
 		},
+		ConnectionLimit: new(int32),
 	}
 
 	query := "SELECT " +
@@ -252,6 +253,8 @@ func (c *external) Observe(ctx context.Context, mg resource.Managed) (managed.Ex
 	}
 	cr.Status.AtProvider.ConfigurationParameters = observed.ConfigurationParameters
 
+	cr.Status.AtProvider.ConnectionLimit = observed.ConnectionLimit
+
 	_, pwdChanged, err := c.getPassword(ctx, cr)
 	if err != nil {
 		return managed.ExternalObservation{}, err
@@ -398,10 +401,11 @@ func (c *external) Update(ctx context.Context, mg resource.Managed) (managed.Ext
 		// Update state to reflect the current configuration parameters
 		cr.Status.AtProvider.ConfigurationParameters = cr.Spec.ForProvider.ConfigurationParameters
 	}
-	cl := cr.Spec.ForProvider.ConnectionLimit
-	if cl != nil {
+	newCl := cr.Spec.ForProvider.ConnectionLimit
+	currCl := cr.Status.AtProvider.ConnectionLimit
+	if (newCl != nil && currCl != nil) && (int64(*currCl) != int64(*newCl)) {
 		if err := c.db.Exec(ctx, xsql.Query{
-			String: fmt.Sprintf("ALTER ROLE %s CONNECTION LIMIT %d", crn, int64(*cl)),
+			String: fmt.Sprintf("ALTER ROLE %s CONNECTION LIMIT %d", crn, int64(*newCl)),
 		}); err != nil {
 			return managed.ExternalUpdate{}, errors.Wrap(err, errUpdateRole)
 		}