Upgrade to modules, pgx/v4 & accoutrements - WIP

Gopkg.toml

@@ -26,7 +26,7 @@
 
 
 [[constraint]]
-  name = "github.com/jackc/pgx"
+  name = "github.com/jackc/pgx/v4"
   branch = "master"
 
 [[constraint]]

config.yaml.sample

@@ -14,10 +14,7 @@ clickhouse:
         debug: false
 
 postgres:
-    host: localhost
-    port: 5432
-    database: pg2ch
-    user: postgres
+    url_or_dsn_connstring: "host=localhost port=5432 dbname=pg2ch user=postgres"
     replication_slot_name: my_slot
     publication_name: my_pub
 

go.mod

@@ -0,0 +1,17 @@
+module github.com/mkabilov/pg2ch
+
+go 1.15
+
+require (
+	github.com/ClickHouse/clickhouse-go v1.4.3
+	github.com/google/btree v1.0.0 // indirect
+	github.com/jackc/pgconn v1.8.0
+	github.com/jackc/pglogrepl v0.0.0-20210109153808-a78a685a0bff
+	github.com/jackc/pgproto3/v2 v2.0.6
+	github.com/jackc/pgtype v1.6.2
+	github.com/jackc/pgx v3.6.2+incompatible
+	github.com/jackc/pgx/v4 v4.10.1
+	github.com/peterbourgon/diskv v2.0.1+incompatible
+	github.com/tidwall/redcon v1.4.0
+	gopkg.in/yaml.v2 v2.4.0
+)

pkg/config/config.go

@@ -8,7 +8,7 @@ import (
 	"strings"
 	"time"
 
-	"github.com/jackc/pgx"
+	"github.com/jackc/pgx/v4"
 	"gopkg.in/yaml.v2"
 
 	"github.com/mkabilov/pg2ch/pkg/message"
@@ -48,8 +48,8 @@ var tableEngines = map[tableEngine]string{
 }
 
 type pgConnConfig struct {
-	pgx.ConnConfig `yaml:",inline"`
-
+	config             *pgx.ConnConfig
+	UrlOrDsnConnstring  string `yaml:"url_or_dsn_connstring"`
 	ReplicationSlotName string `yaml:"replication_slot_name"`
 	PublicationName     string `yaml:"publication_name"`
 }
@@ -214,25 +214,16 @@ func New(filepath string) (*Config, error) {
 		return nil, fmt.Errorf("replication slot name is not specified")
 	}
 
-	connCfg, err := pgx.ParseEnvLibpq()
+	connConfig, err := pgx.ParseConfig(cfg.Postgres.UrlOrDsnConnstring)
 	if err != nil {
-		return nil, fmt.Errorf("could not parse lib pq env variabels: %v", err)
+		return nil, fmt.Errorf("could not parse lib pq env variabels or supplied conn string: %v", err)
 	}
+	cfg.Postgres.config = connConfig
 
 	if cfg.InactivityFlushTimeout.Seconds() == 0 {
 		cfg.InactivityFlushTimeout = defaultInactivityMergeTimeout
 	}
 
-	cfg.Postgres.ConnConfig = cfg.Postgres.ConnConfig.Merge(connCfg)
-
-	if cfg.Postgres.Port == 0 {
-		cfg.Postgres.Port = defaultPostgresPort
-	}
-
-	if cfg.Postgres.Host == "" {
-		cfg.Postgres.Host = defaultPostgresHost
-	}
-
 	if cfg.ClickHouse.Port == 0 {
 		cfg.ClickHouse.Port = defaultClickHousePort
 	}
@@ -296,3 +287,7 @@ func (c *chConnConfig) ConnectionString() string {
 
 	return fmt.Sprintf("tcp://%s:%d?%s", c.Host, c.Port, connStr.Encode())
 }
+
+func (c *Config) PostgresConfig() *(pgx.ConnConfig) {
+	return c.Postgres.config
+}

pkg/consumer/consumer.go

@@ -7,7 +7,9 @@ import (
 	"sync"
 	"time"
 
-	"github.com/jackc/pgx"
+	"github.com/jackc/pgconn"
+	"github.com/jackc/pglogrepl"
+	"github.com/jackc/pgproto3/v2"
 
 	"github.com/mkabilov/pg2ch/pkg/decoder"
 	"github.com/mkabilov/pg2ch/pkg/message"
@@ -35,16 +37,16 @@ type Interface interface {
 type consumer struct {
 	waitGr          *sync.WaitGroup
 	ctx             context.Context
-	conn            *pgx.ReplicationConn
-	dbCfg           pgx.ConnConfig
+	conn            *pgconn.PgConn
+	dbCfg           pgconn.Config
 	slotName        string
 	publicationName string
 	currentLSN      utils.LSN
 	errCh           chan error
 }
 
 // New instantiates the consumer
-func New(ctx context.Context, errCh chan error, dbCfg pgx.ConnConfig, slotName, publicationName string, startLSN utils.LSN) *consumer {
+func New(ctx context.Context, errCh chan error, dbCfg pgconn.Config, slotName, publicationName string, startLSN utils.LSN) *consumer {
 	return &consumer{
 		waitGr:          &sync.WaitGroup{},
 		ctx:             ctx,
@@ -75,7 +77,7 @@ func (c *consumer) close(err error) {
 
 // Run runs consumer
 func (c *consumer) Run(handler Handler) error {
-	rc, err := pgx.ReplicationConnect(c.dbCfg)
+	rc, err := pgconn.ConnectConfig(c.ctx, &c.dbCfg)
 	if err != nil {
 		return fmt.Errorf("could not connect using replication protocol: %v", err)
 	}
@@ -100,8 +102,13 @@ func (c *consumer) Run(handler Handler) error {
 func (c *consumer) startDecoding() error {
 	log.Printf("Starting from %s lsn", c.currentLSN)
 
-	err := c.conn.StartReplication(c.slotName, uint64(c.currentLSN), -1,
-		`"proto_version" '1'`, fmt.Sprintf(`"publication_names" '%s'`, c.publicationName))
+	startReplOptions := pglogrepl.StartReplicationOptions{
+		Timeline: -1,
+		Mode: pglogrepl.LogicalReplication,
+		PluginArgs: []string{`"proto_version" '1'`, fmt.Sprintf(`"publication_names" '%s'`, c.publicationName)},
+	}
+
+	err := pglogrepl.StartReplication(c.ctx, c.conn, c.slotName, pglogrepl.LSN(uint64(c.currentLSN)), startReplOptions)
 
 	if err != nil {
 		c.closeDbConnection()
@@ -112,7 +119,7 @@ func (c *consumer) startDecoding() error {
 }
 
 func (c *consumer) closeDbConnection() {
-	if err := c.conn.Close(); err != nil {
+	if err := c.conn.Close(c.ctx); err != nil {
 		log.Printf("could not close replication connection: %v", err)
 	}
 }
@@ -121,6 +128,7 @@ func (c *consumer) processReplicationMessage(handler Handler) {
 	defer c.waitGr.Done()
 
 	statusTicker := time.NewTicker(statusTimeout)
+
 	for {
 		select {
 		case <-c.ctx.Done():
@@ -134,7 +142,7 @@ func (c *consumer) processReplicationMessage(handler Handler) {
 			}
 		default:
 			wctx, cancel := context.WithTimeout(c.ctx, replWaitTimeout)
-			repMsg, err := c.conn.WaitForReplicationMessage(wctx)
+			repMsg, err := c.conn.ReceiveMessage(wctx)
 			cancel()
 
 			if err == context.DeadlineExceeded {
@@ -153,40 +161,63 @@ func (c *consumer) processReplicationMessage(handler Handler) {
 				continue
 			}
 
-			if repMsg.WalMessage != nil {
-				msg, err := decoder.Parse(repMsg.WalMessage.WalData)
-				if err != nil {
-					c.close(fmt.Errorf("invalid pgoutput message: %s", err))
-					return
-				}
-
-				if err := handler.HandleMessage(utils.LSN(repMsg.WalMessage.WalStart), msg); err != nil {
-					c.close(fmt.Errorf("error handling waldata: %s", err))
-					return
-				}
-			}
-
-			if repMsg.ServerHeartbeat != nil && repMsg.ServerHeartbeat.ReplyRequested == 1 {
-				log.Println("server wants a reply")
-				if err := c.SendStatus(); err != nil {
-					c.close(fmt.Errorf("could not send replay progress: %v", err))
-					return
+			//NEW
+			switch repMsg := repMsg.(type) {
+				case *pgproto3.CopyData:
+					switch repMsg.Data[0] {
+					case pglogrepl.PrimaryKeepaliveMessageByteID:
+						pkm, err := pglogrepl.ParsePrimaryKeepaliveMessage(repMsg.Data[1:])
+						if err != nil {
+							log.Fatalln("ParsePrimaryKeepaliveMessage failed:", err)
+						}
+						log.Println("Primary Keepalive Message =>", "ServerWALEnd:", pkm.ServerWALEnd, "ServerTime:", pkm.ServerTime, "ReplyRequested:", pkm.ReplyRequested)
+
+						//if pkm.ReplyRequested {
+						//	nextStandbyMessageDeadline = time.Time{}
+						//}
+
+					case pglogrepl.XLogDataByteID:
+						xld, err := pglogrepl.ParseXLogData(repMsg.Data[1:])
+						if err != nil {
+							log.Fatalln("ParseXLogData failed:", err)
+						}
+						log.Println("XLogData =>", "WALStart", xld.WALStart, "ServerWALEnd", xld.ServerWALEnd, "ServerTime:", xld.ServerTime, "WALData size", len(xld.WALData))
+
+						//MAYBE clientXLogPos = xld.WALStart + pglogrepl.LSN(len(xld.WALData))
+						msg, err := decoder.Parse(xld.WALData)
+						if err != nil {
+							c.close(fmt.Errorf("invalid pgoutput message: %s", err))
+							return
+						}
+
+						if err := handler.HandleMessage(utils.LSN(xld.WALStart), msg); err != nil {
+							c.close(fmt.Errorf("error handling waldata: %s", err))
+							return
+						}
+
+					}
+				default:
+					log.Printf("Received unexpected message: %#v\n", repMsg)
 				}
-			}
+			// END NEW
+
+			//if repMsg.ServerHeartbeat != nil && repMsg.ServerHeartbeat.ReplyRequested == 1 {
+			//	log.Println("server wants a reply")
+			//	if err := c.SendStatus(); err != nil {
+			//		c.close(fmt.Errorf("could not send replay progress: %v", err))
+			//		return
+			//	}
+			//}
 		}
 	}
 }
 
 // SendStatus sends the status
 func (c *consumer) SendStatus() error {
 	// log.Printf("sending status: %v", c.currentLSN) //TODO: move to debug log level
-	status, err := pgx.NewStandbyStatus(uint64(c.currentLSN))
-
-	if err != nil {
-		return fmt.Errorf("error creating standby status: %s", err)
-	}
+	status := pglogrepl.StandbyStatusUpdate{WALWritePosition: pglogrepl.LSN(uint64(c.currentLSN))}
 
-	if err := c.conn.SendStandbyStatus(status); err != nil {
+	if err := pglogrepl.SendStandbyStatusUpdate(c.ctx, c.conn, status); err != nil {
 		return fmt.Errorf("failed to send standy status: %s", err)
 	}
 

pkg/message/message.go

@@ -5,8 +5,8 @@ import (
 	"strings"
 	"time"
 
-	"github.com/jackc/pgx"
-	"github.com/jackc/pgx/pgtype"
+	"github.com/jackc/pgx/v4"
+	"github.com/jackc/pgtype"
 
 	"github.com/mkabilov/pg2ch/pkg/utils"
 )