feat: add fallback RO db to RW (bxcodec#27)

* chore: backup code

* chore: go-sg-demo

* chore: add DB connection check and fallback from RO to RW

* chore: add fallback logic

* chore: fix linter

* chore: nolint

* chore: remove unnecessary log

Author: bxcodec

.golangci.yaml

@@ -145,4 +145,4 @@ issues:
 run:
   timeout: 5m
   go: "1.19"
-  skip-dirs: []
+  skip-dirs: [examples]

db.go

@@ -50,6 +50,7 @@ type StmtLoadBalancer LoadBalancer[*sql.Stmt]
 // sqlDB is a logical database with multiple underlying physical databases
 // forming a single ReadWrite (primary) with multiple ReadOnly(replicas) db.
 // Reads and writes are automatically directed to the correct db connection
+
 type sqlDB struct {
 	primaries        []*sql.DB
 	replicas         []*sql.DB
@@ -143,18 +144,23 @@ func (db *sqlDB) Prepare(query string) (_stmt Stmt, err error) {
 func (db *sqlDB) PrepareContext(ctx context.Context, query string) (_stmt Stmt, err error) {
 	roStmts := make([]*sql.Stmt, len(db.replicas))
 	primaryStmts := make([]*sql.Stmt, len(db.primaries))
-
 	errPrimaries := doParallely(len(db.primaries), func(i int) (err error) {
 		primaryStmts[i], err = db.primaries[i].PrepareContext(ctx, query)
 		return
 	})
+
 	errReplicas := doParallely(len(db.replicas), func(i int) (err error) {
 		roStmts[i], err = db.replicas[i].PrepareContext(ctx, query)
+		// if connection error happens on RO connection,
+		// ignore and fallback to RW connection
+		if isDBConnectionError(err) {
+			roStmts[i] = primaryStmts[0]
+			return nil
+		}
 		return err
 	})
 
 	err = multierr.Combine(errPrimaries, errReplicas)
-
 	if err != nil {
 		return
 	}
@@ -165,7 +171,7 @@ func (db *sqlDB) PrepareContext(ctx context.Context, query string) (_stmt Stmt,
 		primaryStmts: primaryStmts,
 		replicaStmts: roStmts,
 	}
-	return
+	return _stmt, nil
 }
 
 // Query executes a query that returns rows, typically a SELECT.
@@ -179,7 +185,11 @@ func (db *sqlDB) Query(query string, args ...interface{}) (*sql.Rows, error) {
 // The args are for any placeholder parameters in the query.
 // QueryContext uses a radonly db as the physical db.
 func (db *sqlDB) QueryContext(ctx context.Context, query string, args ...interface{}) (*sql.Rows, error) {
-	return db.ReadOnly().QueryContext(ctx, query, args...)
+	rows, err := db.ReadOnly().QueryContext(ctx, query, args...)
+	if isDBConnectionError(err) {
+		rows, err = db.ReadWrite().QueryContext(ctx, query, args...)
+	}
+	return rows, err
 }
 
 // QueryRow executes a query that is expected to return at most one row.
@@ -195,7 +205,12 @@ func (db *sqlDB) QueryRow(query string, args ...interface{}) *sql.Row {
 // Errors are deferred until Row's Scan method is called.
 // QueryRowContext uses a radonly db as the physical db.
 func (db *sqlDB) QueryRowContext(ctx context.Context, query string, args ...interface{}) *sql.Row {
-	return db.ReadOnly().QueryRowContext(ctx, query, args...)
+	row := db.ReadOnly().QueryRowContext(ctx, query, args...)
+	if isDBConnectionError(row.Err()) {
+		row = db.ReadWrite().QueryRowContext(ctx, query, args...)
+	}
+
+	return row
 }
 
 // SetMaxIdleConns sets the maximum number of connections in the idle

examples/gosg-demo/Makefile

@@ -0,0 +1,72 @@
+# Exporting bin folder to the path for makefile
+export PATH   := $(PWD)/bin:$(PATH)
+# Default Shell
+export SHELL  := bash
+# Type of OS: Linux or Darwin.
+export OSTYPE := $(shell uname -s)
+
+export MUSL   := $(shell [ -x /sbin/apk ] &&  echo "-tags musl" || echo "")
+
+ifeq ($(OSTYPE),Darwin)
+    export MallocNanoZone=0
+endif
+
+define github_url
+    https://github.com/$(GITHUB_REPO)/releases/download/v$(VERSION)/$(ARCHIVE)
+endef
+
+# creates a directory bin.
+bin:
+	@ mkdir -p $@
+
+# ~~~ Tools ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+# ~~ [migrate] ~~~ https://github.com/golang-migrate/migrate ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+MIGRATE := $(shell command -v migrate || echo "bin/migrate")
+migrate: bin/migrate ## Install migrate (database migration)
+
+bin/migrate: VERSION := 4.14.1
+bin/migrate: GITHUB_REPO  := golang-migrate/migrate
+bin/migrate: ARCHIVE := migrate.$(OSTYPE)-amd64.tar.gz
+bin/migrate: bin
+	@ printf "Install migrate... "
+	@ curl -Ls $(call github_url) | tar -zOxf - ./migrate.$(shell echo $(OSTYPE) | tr A-Z a-z)-amd64 > $@ && chmod +x $@
+	@ echo "done."
+
+
+# ~~~ Database Migrations ~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
+
+POSTGRES_USER ?= postgres
+POSTGRES_PASSWORD ?= my_password
+POSTGRES_HOST ?= localhost
+POSTGRES_PORT ?= 5432
+POSTGRES_DATABASE ?= my_database
+
+
+PG_DSN := "postgres://$(POSTGRES_USER):$(POSTGRES_PASSWORD)@$(POSTGRES_HOST):$(POSTGRES_PORT)/$(POSTGRES_DATABASE)?sslmode=disable"
+
+migrate-up: $(MIGRATE) ## Apply all (or N up) migrations.
+	@ read -p "How many migration you wants to perform (default value: [all]): " N; \
+	migrate  -database $(PG_DSN) -path=internal/postgres/migrations up $${N}
+# if you encounter the dirty version, fix the error, then use the below command
+# migrate -database $(PG_DSN) -path=internal/postgres/migrations force previous_version up
+
+.PHONY: migrate-down
+migrate-down: $(MIGRATE) ## Apply all (or N down) migrations.
+	@ read -p "How many migration you wants to perform (default value: [all]): " N; \
+	migrate  -database $(PG_DSN) -path=internal/postgres/migrations down $${N}
+
+.PHONY: migrate-drop
+migrate-drop: $(MIGRATE) ## Drop everything inside the database.
+	migrate  -database $(PG_DSN) -path=internal/postgres/migrations drop
+
+.PHONY: migrate-create
+migrate-create: $(MIGRATE) ## Create a set of up/down migrations with a specified name.
+	@ read -p "Please provide name for the migration: " Name; \
+	migrate create -ext sql -dir internal/postgres/migrations $${Name}
+
+up:
+	@ docker-compose up -d
+down:
+	@ docker-compose up -d

examples/gosg-demo/docker-compose.yaml

@@ -0,0 +1,39 @@
+version: "2"
+
+services:
+  postgres-rw:
+    image: "docker.io/bitnami/postgresql:11-debian-10"
+    ports:
+      - "5432:5432"
+    volumes:
+      - "postgresql_master_data:/bitnami/postgresql"
+    environment:
+      - POSTGRESQL_PGAUDIT_LOG=READ,WRITE
+      - POSTGRESQL_LOG_HOSTNAME=true
+      - POSTGRESQL_REPLICATION_MODE=master
+      - POSTGRESQL_REPLICATION_USER=repl_user
+      - POSTGRESQL_REPLICATION_PASSWORD=repl_password
+      - POSTGRESQL_USERNAME=postgres
+      - POSTGRESQL_DATABASE=my_database
+      # - ALLOW_EMPTY_PASSWORD=yes
+      - POSTGRESQL_PASSWORD=my_password
+  postgres-ro:
+    image: "docker.io/bitnami/postgresql:11-debian-10"
+    ports:
+      - "5433:5432"
+    depends_on:
+      - postgres-rw
+    environment:
+      - POSTGRESQL_USERNAME=postgres
+      - POSTGRESQL_PASSWORD=my_password
+      - POSTGRESQL_MASTER_HOST=postgres-rw
+      - POSTGRESQL_PGAUDIT_LOG=READ,WRITE
+      - POSTGRESQL_LOG_HOSTNAME=true
+      - POSTGRESQL_REPLICATION_MODE=slave
+      - POSTGRESQL_REPLICATION_USER=repl_user
+      - POSTGRESQL_REPLICATION_PASSWORD=repl_password
+      - POSTGRESQL_MASTER_PORT_NUMBER=5432
+
+volumes:
+  postgresql_master_data:
+    driver: local

examples/gosg-demo/internal/postgres/migrations/20230405142308_create_article_table.down.sql

@@ -0,0 +1,5 @@
+BEGIN;
+
+DROP TABLE IF EXISTS articles;
+
+END;

examples/gosg-demo/internal/postgres/migrations/20230405142308_create_article_table.up.sql

@@ -0,0 +1,10 @@
+BEGIN;
+
+CREATE TABLE IF NOT EXISTS articles (
+	article_id serial PRIMARY KEY,
+	title VARCHAR ( 150 ) NOT NULL,
+	content TEXT NOT NULL,
+	created_time TIMESTAMP NOT NULL
+);
+
+END;