Wait for PG to be up before monitoring. Also removing some logging th…

…at is unnecessary

cmd/monitor/main.go

@@ -35,8 +35,11 @@ func main() {
 		panic(fmt.Sprintf("failed to reference node: %s\n", err))
 	}
 
+	// Wait for postgres to boot and become accessible.
+	log.Println("Waiting for Postgres to be ready...")
+	waitOnPostgres(ctx, node)
+
 	// Dead member monitor
-	log.Println("Monitoring dead members")
 	go func() {
 		if err := monitorDeadMembers(ctx, node); err != nil {
 			panic(err)
@@ -59,19 +62,41 @@ func main() {
 		}
 
 		// Backup scheduler
-		log.Println("Monitoring backup schedule")
-		go monitorBackupSchedule(ctx, barman)
+\		go monitorBackupSchedule(ctx, barman)
 
 		// Backup retention monitor
-		log.Println("Monitoring backup retention")
 		go monitorBackupRetention(ctx, barman)
 	}
 
 	// Readonly monitor
-	log.Println("Monitoring cluster state")
 	go monitorClusterState(ctx, node)
 
 	// Replication slot monitor
-	log.Println("Monitoring replication slots")
 	monitorReplicationSlots(ctx, node)
 }
+
+func waitOnPostgres(ctx context.Context, node *flypg.Node) {
+	ticker := time.NewTicker(5 * time.Second)
+	defer ticker.Stop()
+
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			conn, err := node.NewLocalConnection(ctx, "postgres", node.SUCredentials)
+			if err != nil {
+				log.Printf("failed to open local connection: %s", err)
+				continue
+			}
+			defer func() { _ = conn.Close(ctx) }()
+
+			if err := conn.Ping(ctx); err != nil {
+				log.Printf("failed to ping local connection: %s", err)
+				continue
+			}
+
+			return
+		}
+	}
+}

cmd/monitor/monitor_backup_schedule.go

@@ -16,8 +16,6 @@ func monitorBackupSchedule(ctx context.Context, barman *flypg.Barman) {
 		log.Printf("Failed to resolve the last backup taken: %s", err)
 	}
 
-	// TODO - Wait for Postgres to be ready before proceeding.
-
 	// Ensure we have a least one backup before proceeding.
 	if lastBackupTime.IsZero() {
 		log.Println("No backups found! Performing the initial base backup.")
@@ -30,8 +28,6 @@ func monitorBackupSchedule(ctx context.Context, barman *flypg.Barman) {
 		lastBackupTime = time.Now()
 	}
 
-	log.Printf("Last backup taken at: %s", lastBackupTime)
-
 	// Calculate the time until the next backup is due.
 	timeUntilNextBackup := time.Until(lastBackupTime.Add(defaultFullBackupSchedule))
 
@@ -46,7 +42,7 @@ func monitorBackupSchedule(ctx context.Context, barman *flypg.Barman) {
 		timeUntilNextBackup = defaultFullBackupSchedule
 	}
 
-	log.Printf("Next backup due in: %s", timeUntilNextBackup)
+	log.Printf("Next full backup due in: %s", timeUntilNextBackup)
 
 	ticker := time.NewTicker(timeUntilNextBackup)
 	defer ticker.Stop()

internal/utils/shell.go

@@ -13,8 +13,6 @@ import (
 	"syscall"
 )
 
-// TODO - RunCommand needs a context
-
 func RunCmd(ctx context.Context, usr string, name string, args ...string) ([]byte, error) {
 	uid, gid, err := SystemUserIDs(usr)
 	if err != nil {
@@ -27,20 +25,22 @@ func RunCmd(ctx context.Context, usr string, name string, args ...string) ([]byt
 
 	if os.Getenv("DEBUG") != "" {
 		log.Printf("> Running command as %s: %s\n", usr, cmd.String())
-	}
 
-	var stdoutBuf, stderrBuf bytes.Buffer
-	cmd.Stdout = io.MultiWriter(os.Stdout, &stdoutBuf)
-	cmd.Stderr = io.MultiWriter(os.Stderr, &stderrBuf)
+		var stdoutBuf, stderrBuf bytes.Buffer
+		cmd.Stdout = io.MultiWriter(os.Stdout, &stdoutBuf)
+		cmd.Stderr = io.MultiWriter(os.Stderr, &stderrBuf)
 
-	err = cmd.Run()
-	if err != nil {
-		if ee, ok := err.(*exec.ExitError); ok {
-			ee.Stderr = stderrBuf.Bytes()
+		err = cmd.Run()
+		if err != nil {
+			if ee, ok := err.(*exec.ExitError); ok {
+				ee.Stderr = stderrBuf.Bytes()
+			}
 		}
+
+		return stdoutBuf.Bytes(), err
 	}
 
-	return stdoutBuf.Bytes(), err
+	return cmd.Output()
 }
 
 // Deprecated, use RunCmd instead