feat: Aurora MySQL binding to reader endpoint

Aurora has the concept of Writer nodes and Reader nodes. Reader nodes
are useful for High Availability, but they can also be used for
read-only workloads. This features exposes the ability to bind to the
reader endpoint for read-only workloads. This is achieved by passing the
parameters '{"reader":true}' to the 'cf bind-service' command.

[#183390030](https://www.pivotaltracker.com/story/show/183390030)

acceptance-tests/apps/mysqlapp/internal/app/app.go

@@ -4,6 +4,7 @@ import (
 	"database/sql"
 	"fmt"
 	"log"
+	"mysqlapp/internal/connector"
 	"net/http"
 	"time"
 
@@ -17,13 +18,11 @@ const (
 	valueColumn = "valuedata"
 )
 
-func App(config *mysql.Config) *mux.Router {
-	db := connect(config)
-
+func App(conn *connector.Connector) *mux.Router {
 	r := mux.NewRouter()
 	r.HandleFunc("/", aliveness).Methods("HEAD", "GET")
-	r.HandleFunc("/{key}", handleSet(db)).Methods("PUT")
-	r.HandleFunc("/{key}", handleGet(db)).Methods("GET")
+	r.HandleFunc("/{key}", handleSet(conn)).Methods("PUT")
+	r.HandleFunc("/{key}", handleGet(conn)).Methods("GET")
 
 	return r
 }
@@ -33,19 +32,25 @@ func aliveness(w http.ResponseWriter, r *http.Request) {
 	w.WriteHeader(http.StatusNoContent)
 }
 
-func connect(config *mysql.Config) *sql.DB {
+func connect(config *mysql.Config) (*sql.DB, error) {
 	db, err := sql.Open("mysql", config.FormatDSN())
 	if err != nil {
-		log.Fatalf("failed to connect to database: %s", err)
+		return nil, fmt.Errorf("failed to connect to database: %s", err)
 	}
 	db.SetConnMaxLifetime(time.Minute * 3)
 	db.SetMaxOpenConns(10)
 	db.SetMaxIdleConns(10)
 
 	_, err = db.Exec(fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s (%s VARCHAR(255) NOT NULL, %s VARCHAR(255) NOT NULL)`, tableName, keyColumn, valueColumn))
 	if err != nil {
-		log.Fatalf("failed to create test table: %s", err)
+		return nil, fmt.Errorf("failed to create test table: %s", err)
 	}
 
-	return db
+	return db, nil
+}
+
+func fail(w http.ResponseWriter, code int, format string, a ...any) {
+	msg := fmt.Sprintf(format, a...)
+	log.Println(msg)
+	http.Error(w, msg, code)
 }

acceptance-tests/apps/mysqlapp/internal/app/get.go

@@ -1,59 +1,57 @@
 package app
 
 import (
-	"database/sql"
 	"fmt"
 	"log"
+	"mysqlapp/internal/connector"
 	"net/http"
 
 	"github.com/gorilla/mux"
 )
 
-func handleGet(db *sql.DB) func(w http.ResponseWriter, r *http.Request) {
+func handleGet(conn *connector.Connector) func(w http.ResponseWriter, r *http.Request) {
 	return func(w http.ResponseWriter, r *http.Request) {
 		log.Println("Handling get.")
 
+		db, err := conn.Connect()
+		if err != nil {
+			fail(w, http.StatusInternalServerError, "error connecting to database: %s", err)
+		}
+
 		key, ok := mux.Vars(r)["key"]
 		if !ok {
-			log.Println("Key missing.")
-			http.Error(w, "Key missing.", http.StatusBadRequest)
+			fail(w, http.StatusBadRequest, "key missing")
 			return
 		}
 
 		stmt, err := db.Prepare(fmt.Sprintf(`SELECT %s from %s WHERE %s = ?`, valueColumn, tableName, keyColumn))
 		if err != nil {
-			log.Printf("Error preparing statement: %s", err)
-			http.Error(w, "Failed to prepare statement.", http.StatusInternalServerError)
+			fail(w, http.StatusInternalServerError, "error preparing statement: %s", err)
 			return
 		}
 		defer stmt.Close()
 
 		rows, err := stmt.Query(key)
 		if err != nil {
-			log.Printf("Error selecting value: %s", err)
-			http.Error(w, "Failed to select value.", http.StatusNotFound)
+			fail(w, http.StatusNotFound, "error selecting value: %s", err)
 			return
 		}
 		defer rows.Close()
 
 		if !rows.Next() {
-			log.Printf("Error finding value: %s", err)
-			http.Error(w, "Failed to find value.", http.StatusNotFound)
+			fail(w, http.StatusNotFound, "error finding value: %s", err)
 			return
 		}
 
 		var value string
 		if err := rows.Scan(&value); err != nil {
-			log.Printf("Error retrieving value: %s", err)
-			http.Error(w, "Failed to retrieve value.", http.StatusNotFound)
+			fail(w, http.StatusNotFound, "error retrieving value: %s", err)
 			return
 		}
 
 		w.WriteHeader(http.StatusOK)
 		w.Header().Set("Content-Type", "text/html")
-		_, err = w.Write([]byte(value))
-
-		if err != nil {
+		if _, err := w.Write([]byte(value)); err != nil {
 			log.Printf("Error writing value: %s", err)
 			return
 		}

acceptance-tests/apps/mysqlapp/internal/app/set.go

@@ -1,45 +1,51 @@
 package app
 
 import (
-	"database/sql"
 	"fmt"
 	"io"
 	"log"
+	"mysqlapp/internal/connector"
 	"net/http"
 
 	"github.com/gorilla/mux"
 )
 
-func handleSet(db *sql.DB) func(w http.ResponseWriter, r *http.Request) {
+func handleSet(conn *connector.Connector) func(w http.ResponseWriter, r *http.Request) {
 	return func(w http.ResponseWriter, r *http.Request) {
 		log.Println("Handling set.")
 
+		db, err := conn.Connect()
+		if err != nil {
+			fail(w, http.StatusInternalServerError, "error connecting to database: %s", err)
+		}
+
+		_, err = db.Exec(fmt.Sprintf(`CREATE TABLE IF NOT EXISTS %s (%s VARCHAR(255) NOT NULL, %s VARCHAR(255) NOT NULL)`, tableName, keyColumn, valueColumn))
+		if err != nil {
+			fail(w, http.StatusInternalServerError, "failed to create test table: %s", err)
+			return
+		}
+
 		key, ok := mux.Vars(r)["key"]
 		if !ok {
-			log.Println("Key missing.")
-			http.Error(w, "Key missing.", http.StatusBadRequest)
+			fail(w, http.StatusBadRequest, "key missing")
 			return
 		}
 
 		rawValue, err := io.ReadAll(r.Body)
 		if err != nil {
-			log.Printf("Error parsing value: %s", err)
-			http.Error(w, "Failed to parse value.", http.StatusBadRequest)
+			fail(w, http.StatusBadRequest, "error parsing value: %s", err)
 			return
 		}
 
 		stmt, err := db.Prepare(fmt.Sprintf(`INSERT INTO %s (%s, %s) VALUES (?, ?)`, tableName, keyColumn, valueColumn))
 		if err != nil {
-			log.Printf("Error preparing statement: %s", err)
-			http.Error(w, "Failed to prepare statement.", http.StatusInternalServerError)
+			fail(w, http.StatusInternalServerError, "error preparing statement: %s", err)
 			return
 		}
 		defer stmt.Close()
 
-		_, err = stmt.Exec(key, string(rawValue))
-		if err != nil {
-			log.Printf("Error inserting values: %s", err)
-			http.Error(w, "Failed to insert values.", http.StatusBadRequest)
+		if _, err := stmt.Exec(key, string(rawValue)); err != nil {
+			fail(w, http.StatusBadRequest, "error inserting values: %s", err)
 			return
 		}
 

acceptance-tests/apps/mysqlapp/internal/connector/connect.go

@@ -0,0 +1,53 @@
+package connector
+
+import (
+	"database/sql"
+	"fmt"
+	"time"
+
+	"github.com/go-sql-driver/mysql"
+)
+
+type option func(*Connector, *mysql.Config) error
+
+func (c *Connector) Connect(opts ...option) (*sql.DB, error) {
+	cfg := mysql.NewConfig()
+	cfg.Net = "tcp"
+	cfg.Addr = c.Host
+	cfg.User = c.Username
+	cfg.Passwd = c.Password
+	cfg.DBName = c.Database
+	withDefaults(opts...)(c, cfg)
+
+	db, err := sql.Open("mysql", cfg.FormatDSN())
+	if err != nil {
+		return nil, fmt.Errorf("failed to connect to database: %s", err)
+	}
+	db.SetConnMaxLifetime(time.Minute * 3)
+	db.SetMaxOpenConns(10)
+	db.SetMaxIdleConns(10)
+
+	return db, nil
+}
+
+func withOptions(opts ...option) option {
+	return func(conn *Connector, cfg *mysql.Config) error {
+		for _, o := range opts {
+			if err := o(conn, cfg); err != nil {
+				return err
+			}
+		}
+		return nil
+	}
+}
+
+func withDefaults(opts ...option) option {
+	return withOptions(append([]option{withTLS()}, opts...)...)
+}
+
+func withTLS() option {
+	return func(_ *Connector, cfg *mysql.Config) error {
+		cfg.TLSConfig = "true"
+		return nil
+	}
+}

acceptance-tests/apps/mysqlapp/internal/connector/connector.go

@@ -0,0 +1,55 @@
+package connector
+
+import (
+	"fmt"
+
+	"github.com/cloudfoundry-community/go-cfenv"
+	"github.com/mitchellh/mapstructure"
+)
+
+type Connector struct {
+	Host     string `mapstructure:"hostname"`
+	Database string `mapstructure:"name"`
+	Username string `mapstructure:"username"`
+	Password string `mapstructure:"password"`
+	Port     int    `mapstructure:"port"`
+}
+
+func New() (*Connector, error) {
+	app, err := cfenv.Current()
+	if err != nil {
+		return nil, fmt.Errorf("error reading app env: %w", err)
+	}
+	svs, err := app.Services.WithTag("mysql")
+	if err != nil {
+		return nil, fmt.Errorf("error reading MySQL service details")
+	}
+
+	var c Connector
+	if err := mapstructure.Decode(svs[0].Credentials, &c); err != nil {
+		return nil, fmt.Errorf("failed to decode credentials: %w", err)
+	}
+
+	if err := c.Valid(); err != nil {
+		return nil, err
+	}
+
+	return &c, nil
+}
+
+func (c *Connector) Valid() error {
+	switch {
+	case c.Host == "":
+		return fmt.Errorf("missing hostname")
+	case c.Username == "":
+		return fmt.Errorf("missing username")
+	case c.Password == "":
+		return fmt.Errorf("missing password")
+	case c.Database == "":
+		return fmt.Errorf("missing database name")
+	case c.Port == 0:
+		return fmt.Errorf("missing port")
+	}
+
+	return nil
+}

acceptance-tests/apps/mysqlapp/main.go

@@ -4,7 +4,7 @@ import (
 	"fmt"
 	"log"
 	"mysqlapp/internal/app"
-	"mysqlapp/internal/credentials"
+	"mysqlapp/internal/connector"
 	"net/http"
 	"os"
 )
@@ -13,14 +13,14 @@ func main() {
 	log.Println("Starting.")
 
 	log.Println("Reading credentials.")
-	creds, err := credentials.Read()
+	conn, err := connector.New()
 	if err != nil {
 		panic(err)
 	}
 
 	port := port()
 	log.Printf("Listening on port: %s", port)
-	http.Handle("/", app.App(creds))
+	http.Handle("/", app.App(conn))
 	http.ListenAndServe(port, nil)
 }