Merge pull request #107 from fly-apps/remove_pgbouncer

Remove pgbouncer in favor of haproxy for base setup.

Author: davissp14

Dockerfile

@@ -31,7 +31,7 @@ LABEL fly.pg-version=${PG_VERSION}
 LABEL fly.pg-manager=repmgr
 
 RUN apt-get update && apt-get install --no-install-recommends -y \
-    ca-certificates iproute2 postgresql-$PG_MAJOR_VERSION-repmgr curl bash dnsutils vim procps jq pgbouncer ssh \
+    ca-certificates iproute2 postgresql-$PG_MAJOR_VERSION-repmgr curl bash dnsutils vim haproxy socat procps ssh \
     && apt autoremove -y
 
 RUN apt-get update && apt-get install --no-install-recommends -y \
@@ -47,5 +47,4 @@ RUN usermod -d /data postgres
 
 EXPOSE 5432
 
-
 CMD ["start"]

Dockerfile-timescaledb

@@ -31,7 +31,7 @@ LABEL fly.pg-version=${PG_VERSION}
 LABEL fly.pg-manager=repmgr
 
 RUN apt-get update && apt-get install --no-install-recommends -y \
-    ca-certificates iproute2 postgresql-$PG_MAJOR_VERSION-repmgr curl bash dnsutils vim procps jq pgbouncer ssh \
+    ca-certificates iproute2 postgresql-$PG_MAJOR_VERSION-repmgr curl bash dnsutils vim haproxy socat procps ssh \
     && apt autoremove -y
 
 RUN echo "deb https://packagecloud.io/timescale/timescaledb/debian/ $(cat /etc/os-release | grep VERSION_CODENAME | cut -d'=' -f2) main" > /etc/apt/sources.list.d/timescaledb.list \

bin/restart-haproxy

@@ -0,0 +1,4 @@
+#!/bin/bash
+
+PID=$(echo "show info" | socat /var/run/haproxy/haproxy.sock stdio | grep Pid | awk  '{print $2}')
+kill -SIGTERM $PID

cmd/event_handler/main.go

@@ -7,8 +7,6 @@ import (
 	"fmt"
 	"log"
 	"os"
-	"strconv"
-	"time"
 
 	"github.com/fly-apps/postgres-flex/internal/flypg"
 	"github.com/jackc/pgx/v5"
@@ -23,7 +21,6 @@ func main() {
 	// old primary. In the events that we subscribe to it's always either empty or the new primary.
 	// In the future if we subscribe to repmgrd_failover_promote, then we would have to change this
 	// name.
-	newPrimary := flag.String("new-node-id", "", "the new primary node id")
 	success := flag.String("success", "", "success (1) failure (0)")
 	details := flag.String("details", "", "details")
 	flag.Parse()
@@ -39,69 +36,14 @@ func main() {
 	log.SetOutput(logFile)
 	log.Printf("event: %s, node: %d, success: %s, details: %s\n", *event, *nodeID, *success, *details)
 
-	switch *event {
-	case "repmgrd_failover_promote", "standby_promote":
-		// TODO - Need to figure out what to do when success == 0.
-		retry := 0
-		maxRetries := 5
-		success := false
-
-		for retry < maxRetries {
-			if err := reconfigurePGBouncer(*nodeID); err != nil {
-				log.Printf("%s - failed to reconfigure pgbouncer: %s. (attempt: %d)\n", *event, err, retry)
-				retry++
-				time.Sleep(1 * time.Second)
-				continue
-			}
-			success = true
-			break
-		}
-
-		if success {
-			log.Printf("%s - successfully reconfigured pgbouncer to target: %d\n", *event, *nodeID)
-			os.Exit(0)
-		} else {
-			log.Printf("%s - failed to reconfigured pgbouncer to target: %d\n", *event, *nodeID)
-			os.Exit(1)
-		}
-
-	case "standby_follow":
-		newMemberID, err := strconv.Atoi(*newPrimary)
-		if err != nil {
-			log.Printf("failed to parse newMemberID %s: %s\n", *newPrimary, err)
-			os.Exit(1)
-		}
-
-		retry := 0
-		maxRetries := 5
-		success := false
-
-		for retry < maxRetries {
-			if err := reconfigurePGBouncer(newMemberID); err != nil {
-				log.Printf("%s - failed to reconfigure pgbouncer: %s. (attempt: %d)\n", *event, err, retry)
-				retry++
-				time.Sleep(1 * time.Second)
-				continue
-			}
-			success = true
-			break
-		}
-
-		if success {
-			log.Printf("%s - successfully reconfigured pgbouncer to target: %d\n", *event, newMemberID)
-			os.Exit(0)
-		} else {
-			log.Printf("%s - failed to reconfigured pgbouncer to target: %d\n", *event, newMemberID)
-			os.Exit(1)
-		}
+	node, err := flypg.NewNode()
+	if err != nil {
+		log.Printf("failed to initialize node: %s", err)
+		os.Exit(1)
+	}
 
+	switch *event {
 	case "child_node_disconnect", "child_node_reconnect", "child_node_new_connect":
-		node, err := flypg.NewNode()
-		if err != nil {
-			log.Printf("failed to initialize node: %s", err)
-			os.Exit(1)
-		}
-
 		conn, err := node.RepMgr.NewLocalConnection(ctx)
 		if err != nil {
 			log.Printf("failed to open local connection: %s", err)
@@ -132,29 +74,6 @@ func main() {
 	}
 }
 
-func reconfigurePGBouncer(id int) error {
-	node, err := flypg.NewNode()
-	if err != nil {
-		return fmt.Errorf("failed to reference node: %s", err)
-	}
-
-	conn, err := node.RepMgr.NewLocalConnection(context.TODO())
-	if err != nil {
-		return fmt.Errorf("failed to establish connection with local pg: %s", err)
-	}
-
-	member, err := node.RepMgr.MemberByID(context.TODO(), conn, id)
-	if err != nil {
-		return err
-	}
-
-	if err := node.PGBouncer.ConfigurePrimary(context.TODO(), member.Hostname, true); err != nil {
-		return fmt.Errorf("failed to reconfigure pgbouncer primary %s", err)
-	}
-
-	return nil
-}
-
 func evaluateClusterState(ctx context.Context, conn *pgx.Conn, node *flypg.Node) error {
 	standbys, err := node.RepMgr.StandbyMembers(ctx, conn)
 	if err != nil {
@@ -195,9 +114,5 @@ func evaluateClusterState(ctx context.Context, conn *pgx.Conn, node *flypg.Node)
 		}
 	}
 
-	if err := node.PGBouncer.ConfigurePrimary(ctx, primary, true); err != nil {
-		return fmt.Errorf("failed to reconfigure pgbouncer primary %s", err)
-	}
-
 	return nil
 }

cmd/start/main.go

@@ -48,9 +48,13 @@ func main() {
 
 	svisor.AddProcess("postgres", fmt.Sprintf("gosu postgres postgres -D %s -p %d", node.DataDir, node.Port))
 
-	svisor.AddProcess("pgbouncer", fmt.Sprintf("pgbouncer %s/pgbouncer.ini", node.PGBouncer.ConfigPath),
-		supervisor.WithRestart(0, 1*time.Second),
-	)
+	proxyEnv := map[string]string{
+		"FLY_APP_NAME":      os.Getenv("FLY_APP_NAME"),
+		"PRIMARY_REGION":    os.Getenv("PRIMARY_REGION"),
+		"PG_LISTEN_ADDRESS": node.PrivateIP,
+	}
+	svisor.AddProcess("proxy", "/usr/sbin/haproxy -W -db -f /fly/haproxy.cfg", supervisor.WithEnv(proxyEnv), supervisor.WithRestart(0, 1*time.Second))
+
 	svisor.AddProcess("repmgrd", fmt.Sprintf("gosu postgres repmgrd -f %s --daemonize=false", node.RepMgr.ConfigPath),
 		supervisor.WithRestart(0, 5*time.Second),
 	)

config/haproxy.cfg

@@ -0,0 +1,38 @@
+global
+	maxconn 1000
+    stats socket /run/haproxy/haproxy.sock mode 660 level admin
+    stats timeout 2m # Wait up to 2 minutes for input
+
+defaults
+	log	global
+	mode	tcp
+	retries 2
+	timeout client 30m
+	timeout connect 4s
+	timeout server 30m
+	timeout check 5s
+
+resolvers flydns
+    nameserver dns1 [fdaa::3]:53
+    accepted_payload_size 8192 # allow larger DNS payloads
+
+frontend ft_postgresql
+    mode tcp
+    bind *:5432
+	bind :::5432
+	default_backend bk_db
+
+frontend stats
+    mode http
+    bind :::8404
+    stats enable
+    stats uri /stats
+    stats refresh 10s
+
+backend bk_db
+    balance roundrobin
+    option httpchk GET /flycheck/role
+    http-check expect string primary
+    http-check disable-on-404
+    server-template pg 10 $PRIMARY_REGION.$FLY_APP_NAME.internal:5433 check port 5500 resolvers flydns resolve-prefer ipv6 init-addr none on-marked-down shutdown-sessions
+    server pg [$PG_LISTEN_ADDRESS]:5433 check backup port 5500 on-marked-down shutdown-sessions

internal/api/handle_admin.go

@@ -24,6 +24,19 @@ func handleReadonlyState(w http.ResponseWriter, r *http.Request) {
 	renderJSON(w, res, http.StatusOK)
 }
 
+func handleHaproxyRestart(w http.ResponseWriter, r *http.Request) {
+	if err := flypg.RestartHaproxy(); err != nil {
+		renderErr(w, err)
+		return
+	}
+
+	res := &Response{
+		Result: true,
+	}
+
+	renderJSON(w, res, http.StatusOK)
+}
+
 func handleEnableReadonly(w http.ResponseWriter, r *http.Request) {
 	res := &Response{
 		Result: true,
@@ -251,33 +264,6 @@ func (s *Server) handleViewPostgresSettings(w http.ResponseWriter, r *http.Reque
 	renderJSON(w, resp, http.StatusOK)
 }
 
-func (s *Server) handleViewBouncerSettings(w http.ResponseWriter, r *http.Request) {
-	all, err := s.node.PGBouncer.CurrentConfig()
-	if err != nil {
-		renderErr(w, err)
-		return
-	}
-
-	var in []string
-
-	if err := json.NewDecoder(r.Body).Decode(&in); err != nil {
-		renderErr(w, err)
-		return
-	}
-
-	out := map[string]interface{}{}
-
-	for key := range all {
-		val := all[key]
-		if slices.Contains(in, key) {
-			out[key] = val
-		}
-	}
-
-	resp := &Response{Result: out}
-	renderJSON(w, resp, http.StatusOK)
-}
-
 func (s *Server) handleViewRepmgrSettings(w http.ResponseWriter, r *http.Request) {
 	all, err := s.node.RepMgr.CurrentConfig()
 	if err != nil {

internal/api/handler.go

@@ -48,10 +48,10 @@ func (s *Server) Handler() http.Handler {
 		r.Get("/readonly/enable", handleEnableReadonly)
 		r.Get("/readonly/disable", handleDisableReadonly)
 		r.Get("/readonly/state", handleReadonlyState)
+		r.Get("/haproxy/restart", handleHaproxyRestart)
 
 		r.Get("/role", handleRole)
 		r.Get("/settings/view/postgres", s.handleViewPostgresSettings)
-		r.Get("/settings/view/pgbouncer", s.handleViewBouncerSettings)
 		r.Get("/settings/view/repmgr", s.handleViewRepmgrSettings)
 		r.Post("/settings/update/postgres", s.handleUpdatePostgresSettings)
 		r.Post("/settings/apply", s.handleApplyConfig)

internal/flypg/haproxy.go

@@ -0,0 +1,15 @@
+package flypg
+
+import (
+	"fmt"
+
+	"github.com/fly-apps/postgres-flex/internal/utils"
+)
+
+func RestartHaproxy() error {
+	if err := utils.RunCommand("restart-haproxy", "root"); err != nil {
+		return fmt.Errorf("failed to restart haproxy: %s", err)
+	}
+
+	return nil
+}