clean up the diff

Author: Ben Iofel

cmd/monitor/monitor_dead_members.go

@@ -71,13 +71,13 @@ func deadMemberMonitorTick(ctx context.Context, node *flypg.Node, seenAt map[int
 	}
 
 	for _, voter := range votingMembers {
-		sConn, err := node.RepMgr.NewRemoteConnection(ctx, voter.NodeName)
+		sConn, err := node.RepMgr.NewRemoteConnection(ctx, voter.Hostname)
 		if err != nil {
 			// TODO - Verify the exception that's getting thrown.
 			if time.Since(seenAt[voter.ID]) >= deadMemberRemovalThreshold {
-				log.Printf("Removing dead member: %s\n", voter.NodeName)
+				log.Printf("Removing dead member: %s\n", voter.Hostname)
 				if err := node.RepMgr.UnregisterMember(voter); err != nil {
-					log.Printf("failed to unregister member %s: %v", voter.NodeName, err)
+					log.Printf("failed to unregister member %s: %v", voter.Hostname, err)
 					continue
 				}
 				delete(seenAt, voter.ID)

internal/flypg/node.go

@@ -192,7 +192,7 @@ func (n *Node) Init(ctx context.Context) error {
 					return fmt.Errorf("failed to resolve member over dns: %s", err)
 				}
 
-				if err := n.RepMgr.clonePrimary(cloneTarget.NodeName); err != nil {
+				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 {
 						log.Printf("[ERROR] failed to cleanup postgresql dir after clone error: %s\n", rErr)
@@ -270,11 +270,9 @@ func (n *Node) PostInit(ctx context.Context) error {
 			return fmt.Errorf("failed to resolve member role: %s", err)
 		}
 
-		// Restart repmgrd in the event the IP changes for an already registered node.
+		// Restart repmgrd in the event the machine ID changes for an already registered node.
 		// This can happen if the underlying volume is moved to a different node.
-		// TODO - this isn't an IP anymore
-		//daemonRestartRequired := n.RepMgr.daemonRestartRequired(member)
-		daemonRestartRequired := false
+		daemonRestartRequired := n.RepMgr.daemonRestartRequired(member)
 
 		switch member.Role {
 		case PrimaryRoleName:
@@ -333,7 +331,7 @@ func (n *Node) PostInit(ctx context.Context) error {
 			}
 
 			// Register existing witness to apply any configuration changes.
-			if err := n.RepMgr.registerWitness(primary.NodeName); err != nil {
+			if err := n.RepMgr.registerWitness(primary.Hostname); err != nil {
 				return fmt.Errorf("failed to register existing witness: %s", err)
 			}
 		default:
@@ -415,7 +413,7 @@ func (n *Node) PostInit(ctx context.Context) error {
 					return fmt.Errorf("failed to resolve primary member: %s", err)
 				}
 
-				if err := n.RepMgr.registerWitness(primary.NodeName); err != nil {
+				if err := n.RepMgr.registerWitness(primary.Hostname); err != nil {
 					return fmt.Errorf("failed to register witness: %s", err)
 				}
 			} else {
@@ -546,7 +544,7 @@ func (n *Node) migrateNodeNameIfNeeded(ctx context.Context, repConn *pgx.Conn) e
 		return fmt.Errorf("failed to resolve primary member when updating standby: %s", err)
 	}
 
-	primaryConn, err := n.RepMgr.NewRemoteConnection(ctx, primary.NodeName)
+	primaryConn, err := n.RepMgr.NewRemoteConnection(ctx, primary.Hostname)
 	if err != nil {
 		return fmt.Errorf("failed to establish connection to primary: %s", err)
 	}

internal/flypg/readonly.go

@@ -70,31 +70,31 @@ func BroadcastReadonlyChange(ctx context.Context, n *Node, enabled bool) error {
 
 	for _, member := range members {
 		if member.Role == PrimaryRoleName {
-			endpoint := fmt.Sprintf("http://%s:5500/%s", member.NodeName, target)
+			endpoint := fmt.Sprintf("http://%s:5500/%s", member.Hostname, target)
 			resp, err := http.Get(endpoint)
 			if err != nil {
-				log.Printf("[WARN] Failed to broadcast readonly state change to member %s: %s", member.NodeName, 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 {
-				log.Printf("[WARN] Failed to broadcast readonly state change to member %s: %d\n", member.NodeName, resp.StatusCode)
+				log.Printf("[WARN] Failed to broadcast readonly state change to member %s: %d\n", member.Hostname, resp.StatusCode)
 			}
 		}
 	}
 
 	for _, member := range members {
-		endpoint := fmt.Sprintf("http://%s:5500/%s", member.NodeName, RestartHaproxyEndpoint)
+		endpoint := fmt.Sprintf("http://%s:5500/%s", member.Hostname, RestartHaproxyEndpoint)
 		resp, err := http.Get(endpoint)
 		if err != nil {
-			log.Printf("[WARN] Failed to restart haproxy on member %s: %s", member.NodeName, err)
+			log.Printf("[WARN] Failed to restart haproxy on member %s: %s", member.Hostname, err)
 			continue
 		}
 		defer func() { _ = resp.Body.Close() }()
 
 		if resp.StatusCode > 299 {
-			log.Printf("[WARN] Failed to restart haproxy on member %s: %d\n", member.NodeName, resp.StatusCode)
+			log.Printf("[WARN] Failed to restart haproxy on member %s: %d\n", member.Hostname, resp.StatusCode)
 		}
 	}
 

internal/flypg/repmgr.go

@@ -100,19 +100,7 @@ func (r *RepMgr) NewLocalConnection(ctx context.Context) (*pgx.Conn, error) {
 	return openConnection(ctx, host, r.DatabaseName, r.Credentials)
 }
 
-// target - can be an IP address, machine ID in the current app, or other hostname
-func (r *RepMgr) NewRemoteConnection(ctx context.Context, target string) (*pgx.Conn, error) {
-	var hostname string
-
-	ip := net.ParseIP(target)
-	if ip != nil {
-		hostname = target
-	} else if len(target) == 14 {
-		hostname = fmt.Sprintf("%s.vm.%s.internal", target, r.AppName)
-	} else {
-		hostname = target
-	}
-
+func (r *RepMgr) NewRemoteConnection(ctx context.Context, hostname string) (*pgx.Conn, error) {
 	host := net.JoinHostPort(hostname, strconv.Itoa(r.Port))
 	return openConnection(ctx, host, r.DatabaseName, r.Credentials)
 }
@@ -265,7 +253,7 @@ func (r *RepMgr) registerStandby(restartDaemon bool) error {
 }
 
 func (r *RepMgr) registerWitness(primaryHostname string) error {
-	cmdStr := fmt.Sprintf("repmgr witness register -f %s -h %s -F", r.ConfigPath, primaryHostname) // TODO
+	cmdStr := fmt.Sprintf("repmgr witness register -f %s -h %s -F", r.ConfigPath, primaryHostname)
 	_, err := utils.RunCommand(cmdStr, "postgres")
 
 	return err
@@ -291,7 +279,7 @@ func (*RepMgr) restartDaemon() error {
 }
 
 func (r *RepMgr) daemonRestartRequired(m *Member) bool {
-	return m.NodeName != r.PrivateIP // TODO
+	return m.Hostname != r.MachineID
 }
 
 func (r *RepMgr) unregisterWitness(id int) error {
@@ -301,10 +289,10 @@ func (r *RepMgr) unregisterWitness(id int) error {
 	return err
 }
 
-func (r *RepMgr) rejoinCluster(nodeName string) error {
+func (r *RepMgr) rejoinCluster(hostname string) error {
 	cmdStr := fmt.Sprintf("repmgr -f %s node rejoin -h %s -p %d -U %s -d %s --force-rewind --no-wait",
 		r.ConfigPath,
-		nodeName,
+		hostname,
 		r.Port,
 		r.Credentials.Username,
 		r.DatabaseName,
@@ -341,7 +329,7 @@ func (r *RepMgr) regenReplicationConf(ctx context.Context) error {
 	// TODO: do we need -c?
 	if _, err := utils.RunCmd(ctx, "postgres",
 		"repmgr", "--replication-conf-only",
-		"-h", r.PrivateIP, // TODO: should this be the hostname, or even just localhost
+		"-h", r.PrivateIP,
 		"-p", fmt.Sprint(r.Port),
 		"-d", r.DatabaseName,
 		"-U", r.Credentials.Username,
@@ -354,7 +342,7 @@ func (r *RepMgr) regenReplicationConf(ctx context.Context) error {
 
 type Member struct {
 	ID       int
-	NodeName string
+	Hostname string
 	Active   bool
 	Region   string
 	Role     string
@@ -371,7 +359,7 @@ func (*RepMgr) Members(ctx context.Context, pg *pgx.Conn) ([]Member, error) {
 	var members []Member
 	for rows.Next() {
 		var member Member
-		if err := rows.Scan(&member.ID, &member.NodeName, &member.Region, &member.Active, &member.Role); err != nil {
+		if err := rows.Scan(&member.ID, &member.Hostname, &member.Region, &member.Active, &member.Role); err != nil {
 			return nil, err
 		}
 
@@ -404,7 +392,7 @@ func (r *RepMgr) Member(ctx context.Context, conn *pgx.Conn) (*Member, error) {
 func (*RepMgr) PrimaryMember(ctx context.Context, pg *pgx.Conn) (*Member, error) {
 	var member Member
 	sql := "select node_id, node_name, location, active, type from repmgr.nodes where type = 'primary' and active = true;"
-	err := pg.QueryRow(ctx, sql).Scan(&member.ID, &member.NodeName, &member.Region, &member.Active, &member.Role)
+	err := pg.QueryRow(ctx, sql).Scan(&member.ID, &member.Hostname, &member.Region, &member.Active, &member.Role)
 	if err != nil {
 		return nil, err
 	}
@@ -441,7 +429,7 @@ func (*RepMgr) MemberByID(ctx context.Context, pg *pgx.Conn, id int) (*Member, e
 	var member Member
 	sql := fmt.Sprintf("select node_id, node_name, location, active, type from repmgr.nodes where node_id = %d;", id)
 
-	err := pg.QueryRow(ctx, sql).Scan(&member.ID, &member.NodeName, &member.Region, &member.Active, &member.Role)
+	err := pg.QueryRow(ctx, sql).Scan(&member.ID, &member.Hostname, &member.Region, &member.Active, &member.Role)
 	if err != nil {
 		return nil, err
 	}
@@ -453,7 +441,7 @@ func (*RepMgr) MemberByHostname(ctx context.Context, pg *pgx.Conn, hostname stri
 	var member Member
 	sql := fmt.Sprintf("select node_id, node_name, location, active, type from repmgr.nodes where node_name = '%s';", hostname)
 
-	err := pg.QueryRow(ctx, sql).Scan(&member.ID, &member.NodeName, &member.Region, &member.Active, &member.Role)
+	err := pg.QueryRow(ctx, sql).Scan(&member.ID, &member.Hostname, &member.Region, &member.Active, &member.Role)
 	if err != nil {
 		return nil, err
 	}

internal/flypg/zombie.go

@@ -94,9 +94,9 @@ func TakeDNASample(ctx context.Context, node *Node, standbys []Member) (*DNASamp
 
 	for _, standby := range standbys {
 		// Check for connectivity
-		mConn, err := node.RepMgr.NewRemoteConnection(ctx, standby.NodeName)
+		mConn, err := node.RepMgr.NewRemoteConnection(ctx, standby.Hostname)
 		if err != nil {
-			log.Printf("[WARN] Failed to connect to %s\n", standby.NodeName)
+			log.Printf("[WARN] Failed to connect to %s\n", standby.Hostname)
 			sample.totalInactive++
 			continue
 		}
@@ -105,7 +105,7 @@ func TakeDNASample(ctx context.Context, node *Node, standbys []Member) (*DNASamp
 		// Verify the primary
 		primary, err := node.RepMgr.PrimaryMember(ctx, mConn)
 		if err != nil {
-			log.Printf("[WARN] Failed to resolve primary from standby %s\n", standby.NodeName)
+			log.Printf("[WARN] Failed to resolve primary from standby %s\n", standby.Hostname)
 			sample.totalInactive++
 			continue
 		}
@@ -117,9 +117,9 @@ func TakeDNASample(ctx context.Context, node *Node, standbys []Member) (*DNASamp
 		sample.totalActive++
 
 		// Record conflict when primary doesn't match.
-		if primary.NodeName != node.MachineID && primary.NodeName != node.PrivateIP {
+		if primary.Hostname != node.MachineID && primary.Hostname != node.PrivateIP {
 			sample.totalConflicts++
-			sample.conflictMap[primary.NodeName]++
+			sample.conflictMap[primary.Hostname]++
 		}
 	}
 
@@ -198,11 +198,6 @@ func handleZombieLock(ctx context.Context, n *Node) error {
 	// If the zombie lock contains a hostname, it means we were able to
 	// resolve the real primary and will attempt to rejoin it.
 	if primaryStr != "" {
-		//ip := net.ParseIP(primaryStr)
-		//if ip == nil {
-		//	return fmt.Errorf("zombie.lock file contains an invalid ipv6 address")
-		//}
-
 		conn, err := n.RepMgr.NewRemoteConnection(ctx, primaryStr)
 		if err != nil {
 			return fmt.Errorf("failed to establish a connection to our rejoin target %s: %s", primaryStr, err)
@@ -215,8 +210,7 @@ func handleZombieLock(ctx context.Context, n *Node) error {
 		}
 
 		// Confirm that our rejoin target still identifies itself as the primary.
-		// TODO - machine IDs don't change
-		if primary.NodeName != primaryStr {
+		if primary.Hostname != primaryStr {
 			// Clear the zombie.lock file so we can attempt to re-resolve the correct primary.
 			if err := RemoveZombieLock(); err != nil {
 				return fmt.Errorf("failed to remove zombie lock: %s", err)
@@ -231,7 +225,7 @@ func handleZombieLock(ctx context.Context, n *Node) error {
 			return ErrZombieLockRegionMismatch
 		}
 
-		if err := n.RepMgr.rejoinCluster(primary.NodeName); err != nil {
+		if err := n.RepMgr.rejoinCluster(primary.Hostname); err != nil {
 			return fmt.Errorf("failed to rejoin cluster: %s", err)
 		}