Threadsafe logging (#148)

* Add logging that's threadsafe

* Cleanup

* Removing info tags

* Updates

Author: davissp14

cmd/start/main.go

@@ -3,6 +3,7 @@ package main
 import (
 	"context"
 	"fmt"
+	"log"
 	"os"
 	"syscall"
 	"time"
@@ -12,6 +13,8 @@ import (
 )
 
 func main() {
+	log.SetFlags(0)
+
 	requiredPasswords := []string{"SU_PASSWORD", "OPERATOR_PASSWORD", "REPL_PASSWORD"}
 	for _, str := range requiredPasswords {
 		if _, exists := os.LookupEnv(str); !exists {

internal/api/response.go

@@ -35,7 +35,6 @@ func status(err error) int {
 	var pgErr *pgconn.PgError
 
 	if errors.As(err, &pgErr) {
-		// fmt.Printf("%s: %s\n", pgErr.Code, pgErr.Message)
 		switch pgErr.Code {
 		case "42710": // unique violation
 			return http.StatusConflict

internal/flycheck/pg.go

@@ -3,6 +3,7 @@ package flycheck
 import (
 	"context"
 	"fmt"
+	"log"
 
 	"github.com/fly-apps/postgres-flex/internal/flypg"
 	"github.com/jackc/pgx/v5"
@@ -69,9 +70,9 @@ func diskCapacityCheck(ctx context.Context, node *flypg.Node) (string, error) {
 		// If the read-only lock has already been set, we can assume that we've already
 		// broadcasted.
 		if !flypg.ReadOnlyLockExists() {
-			fmt.Println("Broadcasting readonly change to registered standbys")
+			log.Println("[WARN] Broadcasting readonly change to registered standbys")
 			if err := flypg.BroadcastReadonlyChange(ctx, node, true); err != nil {
-				fmt.Printf("errors with enable readonly broadcast: %s\n", err)
+				log.Printf("[ERROR] Failed to enable readonly: %s\n", err)
 			}
 		}
 
@@ -81,7 +82,7 @@ func diskCapacityCheck(ctx context.Context, node *flypg.Node) (string, error) {
 	// Don't attempt to disable readonly if there's a zombie.lock
 	if !flypg.ZombieLockExists() && flypg.ReadOnlyLockExists() {
 		if err := flypg.BroadcastReadonlyChange(ctx, node, false); err != nil {
-			fmt.Printf("errors with disable readonly broadcast: %s\n", err)
+			log.Printf("[ERROR] Failed to disable readonly: %s\n", err)
 		}
 	}
 

internal/flypg/node.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"log"
 	"net"
 	"os"
 	"os/exec"
@@ -163,7 +164,7 @@ func (n *Node) Init(ctx context.Context) error {
 		}
 
 		if !clusterInitialized {
-			fmt.Println("Provisioning primary")
+			log.Println("Provisioning primary")
 			// TODO - This should probably run on boot in case the password changes.
 			if err := n.PGConfig.writePasswordFile(n.OperatorCredentials.Password); err != nil {
 				return fmt.Errorf("failed to write pg password file: %s", err)
@@ -173,7 +174,7 @@ func (n *Node) Init(ctx context.Context) error {
 				return fmt.Errorf("failed to initialize postgres %s", err)
 			}
 		} else {
-			fmt.Println("Provisioning standby")
+			log.Println("Provisioning standby")
 			cloneTarget, err := n.RepMgr.ResolveMemberOverDNS(ctx)
 			if err != nil {
 				return fmt.Errorf("failed to resolve member over dns: %s", err)
@@ -182,7 +183,7 @@ func (n *Node) Init(ctx context.Context) error {
 			if err := n.RepMgr.clonePrimary(cloneTarget.Hostname); err != nil {
 				// Clean-up the directory so it can be retried.
 				if rErr := os.Remove(n.DataDir); rErr != nil {
-					fmt.Printf("failed to cleanup postgresql dir after clone error: %s\n", rErr)
+					log.Printf("[ERROR] failed to cleanup postgresql dir after clone error: %s\n", rErr)
 				}
 
 				return fmt.Errorf("failed to clone primary: %s", err)
@@ -208,8 +209,8 @@ func (n *Node) Init(ctx context.Context) error {
 // PostInit are operations that need to be executed against a running Postgres on boot.
 func (n *Node) PostInit(ctx context.Context) error {
 	if ZombieLockExists() {
-		fmt.Println("Manual intervention required. Delete the zombie.lock file and restart the machine to force a retry.")
-		fmt.Println("Sleeping for 5 minutes.")
+		log.Println("[ERROR] Manual intervention required. Delete the zombie.lock file and restart the machine to force a retry.")
+		log.Println("[ERROR] Sleeping for 5 minutes.")
 		time.Sleep(5 * time.Minute)
 		return fmt.Errorf("unrecoverable zombie")
 	}
@@ -219,8 +220,8 @@ func (n *Node) PostInit(ctx context.Context) error {
 	if err != nil {
 		// Check to see if this is an authentication error.
 		if strings.Contains(err.Error(), "28P01") {
-			fmt.Println("WARNING: `postgres` user password does not match the `OPERATOR_PASSWORD` secret")
-			fmt.Printf("HINT: Use `fly secrets set OPERATOR_PASSWORD=<password> --app %s` to resolve the issue\n", n.AppName)
+			log.Println("[WARN] `postgres` user password does not match the `OPERATOR_PASSWORD` secret")
+			log.Printf("[WARN] Use `fly secrets set OPERATOR_PASSWORD=<password> --app %s` to resolve the issue\n", n.AppName)
 		}
 
 		return fmt.Errorf("failed to establish connection to local node: %s", err)
@@ -251,13 +252,13 @@ func (n *Node) PostInit(ctx context.Context) error {
 			// Verify cluster state to ensure we are the actual primary and not a zombie.
 			primary, err := PerformScreening(ctx, conn, n)
 			if errors.Is(err, ErrZombieDiagnosisUndecided) {
-				fmt.Println("Unable to confirm that we are the true primary!")
+				log.Println("[ERROR] Unable to confirm that we are the true primary!")
 				// Turn member read-only
 				if err := Quarantine(ctx, n, primary); err != nil {
 					return fmt.Errorf("failed to quarantine failed primary: %s", err)
 				}
 			} else if errors.Is(err, ErrZombieDiscovered) {
-				fmt.Printf("The majority of registered members agree that '%s' is the real primary.\n", primary)
+				log.Printf("[ERROR] The majority of registered members agree that '%s' is the real primary.\n", primary)
 				// Turn member read-only
 				if err := Quarantine(ctx, n, primary); err != nil {
 					return fmt.Errorf("failed to quarantine failed primary: %s", err)
@@ -285,7 +286,7 @@ func (n *Node) PostInit(ctx context.Context) error {
 		case StandbyRoleName:
 			// Register existing standby to take-on any configuration changes.
 			if err := n.RepMgr.registerStandby(); err != nil {
-				fmt.Printf("failed to register existing standby: %s\n", err)
+				return fmt.Errorf("failed to register existing standby: %s", err)
 			}
 		default:
 			return fmt.Errorf("member has unknown role: %q", member.Role)
@@ -312,7 +313,7 @@ func (n *Node) PostInit(ctx context.Context) error {
 
 		if !clusterInitialized {
 			// Configure as primary
-			fmt.Println("Registering primary")
+			log.Println("Registering primary")
 
 			// Verify we reside within the clusters primary region
 			if !n.RepMgr.eligiblePrimary() {
@@ -349,9 +350,9 @@ func (n *Node) PostInit(ctx context.Context) error {
 			}
 		} else {
 			// Configure as standby
-			fmt.Println("Registering standby")
+			log.Println("Registering standby")
 			if err := n.RepMgr.registerStandby(); err != nil {
-				fmt.Printf("failed to register new standby: %s\n", err)
+				return fmt.Errorf("failed to register new standby: %s", err)
 			}
 
 			// Let the boot process know that we've already been configured.

internal/flypg/pg.go

@@ -5,6 +5,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"io"
+	"log"
 	"os"
 	"strconv"
 	"strings"
@@ -177,7 +178,7 @@ func (c *PGConfig) SetDefaults() error {
 func (c *PGConfig) RuntimeApply(ctx context.Context, conn *pgx.Conn) error {
 	for key, value := range c.userConfig {
 		if err := admin.SetConfigurationSetting(ctx, conn, key, value); err != nil {
-			fmt.Printf("failed to set configuration setting %s -> %s: %s", key, value, err)
+			log.Printf("[WARN] Failed to set configuration setting %s -> %s: %s", key, value, err)
 		}
 	}
 
@@ -230,8 +231,8 @@ func (c *PGConfig) initialize(store *state.Store) error {
 	}
 
 	if err := SyncUserConfig(c, store); err != nil {
-		fmt.Printf("WARNING: Failed to sync user config from consul for postgres: %s\n", err.Error())
-		fmt.Println("       This may cause this node to behave unexpectedly")
+		log.Printf("[WARN] Failed to sync user config from consul for postgres: %s\n", err.Error())
+		log.Println("[WARN] This may cause this node to behave unexpectedly")
 		if err := writeInternalConfigFile(c); err != nil {
 			return fmt.Errorf("failed to write pg config files: %s", err)
 		}

internal/flypg/readonly.go

@@ -3,6 +3,7 @@ package flypg
 import (
 	"context"
 	"fmt"
+	"log"
 	"net/http"
 	"os"
 	"time"
@@ -72,13 +73,13 @@ func BroadcastReadonlyChange(ctx context.Context, n *Node, enabled bool) error {
 			endpoint := fmt.Sprintf("http://[%s]:5500/%s", member.Hostname, target)
 			resp, err := http.Get(endpoint)
 			if err != nil {
-				fmt.Printf("failed to broadcast readonly state change to member %s: %s", member.Hostname, err)
+				log.Printf("[WARN] Failed to broadcast readonly state change to member %s: %s", member.Hostname, err)
 				continue
 			}
 			defer func() { _ = resp.Body.Close() }()
 
 			if resp.StatusCode > 299 {
-				fmt.Printf("failed to broadcast readonly state change to member %s: %d\n", member.Hostname, resp.StatusCode)
+				log.Printf("[WARN] Failed to broadcast readonly state change to member %s: %d\n", member.Hostname, resp.StatusCode)
 			}
 		}
 	}
@@ -87,13 +88,13 @@ func BroadcastReadonlyChange(ctx context.Context, n *Node, enabled bool) error {
 		endpoint := fmt.Sprintf("http://[%s]:5500/%s", member.Hostname, RestartHaproxyEndpoint)
 		resp, err := http.Get(endpoint)
 		if err != nil {
-			fmt.Printf("failed restart haproxy on member %s: %s", member.Hostname, err)
+			log.Printf("[WARN] Failed restart haproxy on member %s: %s", member.Hostname, err)
 			continue
 		}
 		defer func() { _ = resp.Body.Close() }()
 
 		if resp.StatusCode > 299 {
-			fmt.Printf("failed to restart haproxy on member %s: %d\n", member.Hostname, resp.StatusCode)
+			log.Printf("[WARN] Failed to restart haproxy on member %s: %d\n", member.Hostname, resp.StatusCode)
 		}
 	}
 
@@ -115,7 +116,7 @@ func writeReadOnlyLock() error {
 	}
 
 	if err := os.WriteFile(readOnlyLockFile, []byte(time.Now().String()), 0600); err != nil {
-		return err
+		return fmt.Errorf("failed to create readonly.lock: %s", err)
 	}
 
 	if err := utils.SetFileOwnership(readOnlyLockFile, "postgres"); err != nil {

internal/flypg/registration.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"log"
 	"os"
 
 	"github.com/fly-apps/postgres-flex/internal/flypg/admin"
@@ -53,7 +54,7 @@ func isRegistered(ctx context.Context, conn *pgx.Conn, n *Node) (bool, error) {
 	// If we are active, issue registration certificate
 	if member.Active {
 		if err := issueRegistrationCert(); err != nil {
-			fmt.Println("failed to issue registration certificate.")
+			log.Println("[WARN] Failed to issue registration certificate.")
 			return true, nil
 		}
 	}

internal/flypg/repmgr.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/rand"
 	"fmt"
+	"log"
 	"math"
 	"math/big"
 	"net"
@@ -235,7 +236,7 @@ func (r *RepMgr) rejoinCluster(hostname string) error {
 		r.DatabaseName,
 	)
 
-	fmt.Println(cmdStr)
+	log.Println(cmdStr)
 	_, err := utils.RunCommand(cmdStr, "postgres")
 
 	return err
@@ -245,7 +246,7 @@ func (r *RepMgr) registerStandby() error {
 	// Force re-registry to ensure the standby picks up any new configuration changes.
 	cmdStr := fmt.Sprintf("repmgr -f %s standby register -F", r.ConfigPath)
 	if _, err := utils.RunCommand(cmdStr, "postgres"); err != nil {
-		fmt.Printf("failed to register standby: %s", err)
+		return err
 	}
 
 	return nil
@@ -254,7 +255,7 @@ func (r *RepMgr) registerStandby() error {
 func (r *RepMgr) unregisterStandby(id int) error {
 	cmdStr := fmt.Sprintf("repmgr standby unregister -f %s --node-id=%d", r.ConfigPath, id)
 	if _, err := utils.RunCommand(cmdStr, "postgres"); err != nil {
-		fmt.Printf("failed to unregister standby: %s", err)
+		return err
 	}
 
 	return nil
@@ -273,7 +274,7 @@ func (r *RepMgr) clonePrimary(ipStr string) error {
 		r.Credentials.Username,
 		r.ConfigPath)
 
-	fmt.Println(cmdStr)
+	log.Println(cmdStr)
 	if _, err := utils.RunCommand(cmdStr, "postgres"); err != nil {
 		return fmt.Errorf("failed to clone primary: %s", err)
 	}
@@ -397,7 +398,6 @@ func (r *RepMgr) ResolveMemberOverDNS(ctx context.Context) (*Member, error) {
 
 		member, err := r.MemberByHostname(ctx, conn, ip.String())
 		if err != nil {
-			fmt.Printf("failed to resolve role from %s", ip.String())
 			continue
 		}
 

internal/flypg/restore.go

@@ -52,7 +52,7 @@ func Restore(ctx context.Context, node *Node) error {
 
 	go func() {
 		if err := svisor.Run(); err != nil {
-			log.Printf("failed to boot postgres in the background: %s", err)
+			log.Printf("[ERROR] failed to boot postgres in the background: %s", err)
 		}
 	}()
 

internal/flypg/zombie.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"errors"
 	"fmt"
+	"log"
 	"net"
 	"os"
 
@@ -68,7 +69,7 @@ func PerformScreening(ctx context.Context, conn *pgx.Conn, n *Node) (string, err
 		return "", fmt.Errorf("failed to evaluate cluster data: %s", err)
 	}
 
-	fmt.Println(DNASampleString(sample))
+	log.Println(DNASampleString(sample))
 
 	return ZombieDiagnosis(sample)
 }
@@ -96,7 +97,7 @@ func TakeDNASample(ctx context.Context, node *Node, standbys []Member) (*DNASamp
 		// Check for connectivity
 		mConn, err := node.RepMgr.NewRemoteConnection(ctx, standby.Hostname)
 		if err != nil {
-			fmt.Printf("failed to connect to %s\n", standby.Hostname)
+			log.Printf("[WARN] Failed to connect to %s\n", standby.Hostname)
 			sample.totalInactive++
 			continue
 		}
@@ -105,7 +106,7 @@ func TakeDNASample(ctx context.Context, node *Node, standbys []Member) (*DNASamp
 		// Verify the primary
 		primary, err := node.RepMgr.PrimaryMember(ctx, mConn)
 		if err != nil {
-			fmt.Printf("failed to resolve primary from standby %s\n", standby.Hostname)
+			log.Printf("[WARN] Failed to resolve primary from standby %s\n", standby.Hostname)
 			sample.totalInactive++
 			continue
 		}
@@ -190,7 +191,7 @@ func DNASampleString(s *DNASample) string {
 }
 
 func handleZombieLock(ctx context.Context, n *Node) error {
-	fmt.Println("Zombie lock detected!")
+	log.Println("[WARN] Zombie lock detected!")
 	primaryStr, err := ReadZombieLock()
 	if err != nil {
 		return fmt.Errorf("failed to read zombie lock: %s", primaryStr)
@@ -227,7 +228,7 @@ func handleZombieLock(ctx context.Context, n *Node) error {
 
 		// If the primary does not reside within our primary region, we cannot rejoin until it is.
 		if primary.Region != n.PrimaryRegion {
-			fmt.Printf("Primary region mismatch detected. The primary lives in '%s', while PRIMARY_REGION is set to '%s'\n", primary.Region, n.PrimaryRegion)
+			log.Printf("[WARN] Primary region mismatch detected. The primary lives in '%s', while PRIMARY_REGION is set to '%s'\n", primary.Region, n.PrimaryRegion)
 			return ErrZombieLockRegionMismatch
 		}
 
@@ -248,8 +249,8 @@ func handleZombieLock(ctx context.Context, n *Node) error {
 		}
 	} else {
 		// TODO - Provide link to documention on how to address this
-		fmt.Println("Zombie lock file does not contain a hostname.")
-		fmt.Println("This likely means that we were unable to determine who the real primary is.")
+		log.Println("[WARN] Zombie lock file does not contain a hostname.")
+		log.Println("[WARN] This likely means that we were unable to determine who the real primary is.")
 	}
 
 	return nil