Replace readOnlyStateMonitor with clusterStateMonitor

Author: davissp14

cmd/event_handler/main.go

@@ -71,27 +71,12 @@ func main() {
 }
 
 func evaluateClusterState(ctx context.Context, conn *pgx.Conn, node *flypg.Node) error {
-	standbys, err := node.RepMgr.StandbyMembers(ctx, conn)
-	if err != nil {
-		if !errors.Is(err, pgx.ErrNoRows) {
-			return fmt.Errorf("failed to query standbys")
-		}
-	}
-
-	sample, err := flypg.TakeDNASample(ctx, node, standbys)
-	if err != nil {
-		return fmt.Errorf("failed to evaluate cluster data: %s", err)
-	}
-
-	log.Println(flypg.DNASampleString(sample))
-
-	primary, err := flypg.ZombieDiagnosis(sample)
+	primary, err := node.EvaluateClusterState(ctx, conn)
 	if errors.Is(err, flypg.ErrZombieDiagnosisUndecided) || errors.Is(err, flypg.ErrZombieDiscovered) {
 		// Quarantine primary
 		if err := flypg.Quarantine(ctx, conn, node, primary); err != nil {
 			return fmt.Errorf("failed to quarantine failed primary: %s", err)
 		}
-
 		return fmt.Errorf("primary has been quarantined: %s", err)
 	} else if err != nil {
 		return fmt.Errorf("failed to run zombie diagnosis: %s", err)

cmd/monitor/main.go

@@ -12,7 +12,7 @@ import (
 var (
 	deadMemberMonitorFrequency       = time.Hour * 1
 	replicationStateMonitorFrequency = time.Hour * 1
-	readonlyStateMonitorFrequency    = time.Minute * 1
+	clusterStateMonitorFrequency     = time.Minute * 15
 
 	defaultDeadMemberRemovalThreshold   = time.Hour * 12
 	defaultInactiveSlotRemovalThreshold = time.Hour * 12
@@ -35,8 +35,8 @@ func main() {
 	}()
 
 	// Readonly monitor
-	log.Println("Monitoring readonly state")
-	go monitorReadOnly(ctx, node)
+	log.Println("Monitoring cluster state")
+	go monitorClusterState(ctx, node)
 
 	// Replication slot monitor
 	log.Println("Monitoring replication slots")

cmd/monitor/monitor_cluster_state.go

@@ -0,0 +1,93 @@
+package main
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"log"
+	"time"
+
+	"github.com/fly-apps/postgres-flex/internal/flypg"
+	"github.com/jackc/pgx/v5"
+)
+
+func monitorClusterState(ctx context.Context, node *flypg.Node) {
+	ticker := time.NewTicker(clusterStateMonitorFrequency)
+	defer ticker.Stop()
+	for range ticker.C {
+		if err := clusterStateMonitorTick(ctx, node); err != nil {
+			log.Printf("clusterStateMonitorTick failed with: %s", err)
+		}
+	}
+}
+
+func clusterStateMonitorTick(ctx context.Context, node *flypg.Node) error {
+	conn, err := node.RepMgr.NewLocalConnection(ctx)
+	if err != nil {
+		return fmt.Errorf("failed to open local connection: %s", err)
+	}
+	defer conn.Close(ctx)
+
+	member, err := node.RepMgr.Member(ctx, conn)
+	if err != nil {
+		return fmt.Errorf("failed to query local member: %s", err)
+	}
+
+	// We only need to monitor the primary
+	if member.Role != flypg.PrimaryRoleName {
+		return nil
+	}
+
+	primary, err := node.EvaluateClusterState(ctx, conn)
+	if errors.Is(err, flypg.ErrZombieDiagnosisUndecided) || errors.Is(err, flypg.ErrZombieDiscovered) {
+		if err := flypg.Quarantine(ctx, conn, node, primary); err != nil {
+			return fmt.Errorf("failed to quarantine failed primary: %s", err)
+		}
+		return fmt.Errorf("primary has been quarantined: %s", err)
+	} else if err != nil {
+		return fmt.Errorf("failed to run zombie diagnosis: %s", err)
+	}
+
+	// Clear zombie lock if it exists
+	if flypg.ZombieLockExists() {
+		log.Println("Clearing zombie lock and enabling read/write")
+		if err := flypg.RemoveZombieLock(); err != nil {
+			return fmt.Errorf("failed to remove zombie lock: %s", err)
+		}
+
+		log.Println("Broadcasting readonly state change")
+		if err := flypg.BroadcastReadonlyChange(ctx, node, false); err != nil {
+			log.Printf("errors while disabling readonly: %s", err)
+		}
+	}
+
+	return nil
+}
+
+func evaluateClusterState(ctx context.Context, conn *pgx.Conn, node *flypg.Node) error {
+	primary, err := node.EvaluateClusterState(ctx, conn)
+	if errors.Is(err, flypg.ErrZombieDiagnosisUndecided) || errors.Is(err, flypg.ErrZombieDiscovered) {
+		// Quarantine primary
+		if err := flypg.Quarantine(ctx, conn, node, primary); err != nil {
+			return fmt.Errorf("failed to quarantine failed primary: %s", err)
+		}
+		return fmt.Errorf("primary has been quarantined: %s", err)
+	} else if err != nil {
+		return fmt.Errorf("failed to run zombie diagnosis: %s", err)
+	}
+
+	// Clear zombie lock if it exists
+	if flypg.ZombieLockExists() {
+		log.Println("Clearing zombie lock and enabling read/write")
+		if err := flypg.RemoveZombieLock(); err != nil {
+			return fmt.Errorf("failed to remove zombie lock: %s", err)
+		}
+
+		log.Println("Broadcasting readonly state change")
+		if err := flypg.BroadcastReadonlyChange(ctx, node, false); err != nil {
+			log.Printf("errors while disabling readonly: %s", err)
+		}
+	}
+
+	return nil
+}

cmd/monitor/monitor_readonly.go

@@ -326,30 +326,12 @@ func (n *Node) PostInit(ctx context.Context) error {
 
 		switch role {
 		case PrimaryRoleName:
-			standbys, err := repmgr.StandbyMembers(ctx, conn)
-			if err != nil {
-				if !errors.Is(err, pgx.ErrNoRows) {
-					return fmt.Errorf("failed to query standbys")
-				}
-			}
-
-			// Collect sample data from registered standbys
-			sample, err := TakeDNASample(ctx, n, standbys)
-			if err != nil {
-				return fmt.Errorf("failed to resolve cluster metrics: %s", err)
-			}
-
-			fmt.Println(DNASampleString(sample))
-
-			// Evaluate whether we are a zombie or not.
-			primary, err := ZombieDiagnosis(sample)
+			primary, err := n.EvaluateClusterState(ctx, conn)
 			if errors.Is(err, ErrZombieDiagnosisUndecided) {
 				fmt.Println("Unable to confirm that we are the true primary!")
-
 				if err := Quarantine(ctx, conn, 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)
 
@@ -678,3 +660,21 @@ func setDirOwnership() error {
 	_, err = cmd.Output()
 	return err
 }
+
+func (n *Node) EvaluateClusterState(ctx context.Context, conn *pgx.Conn) (string, error) {
+	standbys, err := n.RepMgr.StandbyMembers(ctx, conn)
+	if err != nil {
+		if !errors.Is(err, pgx.ErrNoRows) {
+			return "", fmt.Errorf("failed to query standbys")
+		}
+	}
+
+	sample, err := TakeDNASample(ctx, n, standbys)
+	if err != nil {
+		return "", fmt.Errorf("failed to evaluate cluster data: %s", err)
+	}
+
+	fmt.Println(DNASampleString(sample))
+
+	return ZombieDiagnosis(sample)
+}

internal/flypg/node.go

@@ -157,6 +157,10 @@ func ZombieDiagnosis(s *DNASample) (string, error) {
 }
 
 func Quarantine(ctx context.Context, conn *pgx.Conn, n *Node, primary string) error {
+	if ZombieLockExists() {
+		return nil
+	}
+
 	fmt.Println("Writing zombie.lock file.")
 	if err := writeZombieLock(primary); err != nil {
 		return fmt.Errorf("failed to set zombie lock: %s", err)