feat(db): clickhouse failover support

Author: invis-bitfly

backend/go.mod

@@ -234,6 +234,7 @@ require (
 	github.com/sanity-io/litter v1.5.5 // indirect
 	github.com/segmentio/asm v1.2.0 // indirect
 	github.com/sergi/go-diff v1.2.0 // indirect
+	github.com/sethvargo/go-envconfig v1.1.0 // indirect
 	github.com/sethvargo/go-retry v0.2.4 // indirect
 	github.com/shirou/gopsutil v3.21.11+incompatible // indirect
 	github.com/spaolacci/murmur3 v1.1.0 // indirect

backend/go.sum

@@ -862,6 +862,8 @@ github.com/segmentio/asm v1.2.0/go.mod h1:BqMnlJP91P8d+4ibuonYZw9mfnzI9HfxselHZr
 github.com/sergi/go-diff v1.0.0/go.mod h1:0CfEIISq7TuYL3j771MWULgwwjU+GofnZX9QAmXWZgo=
 github.com/sergi/go-diff v1.2.0 h1:XU+rvMAioB0UC3q1MFrIQy4Vo5/4VsRDQQXHsEya6xQ=
 github.com/sergi/go-diff v1.2.0/go.mod h1:STckp+ISIX8hZLjrqAeVduY0gWCT9IjLuqbuNXdaHfM=
+github.com/sethvargo/go-envconfig v1.1.0 h1:cWZiJxeTm7AlCvzGXrEXaSTCNgip5oJepekh/BOQuog=
+github.com/sethvargo/go-envconfig v1.1.0/go.mod h1:JLd0KFWQYzyENqnEPWWZ49i4vzZo/6nRidxI8YvGiHw=
 github.com/sethvargo/go-retry v0.2.4 h1:T+jHEQy/zKJf5s95UkguisicE0zuF9y7+/vgz08Ocec=
 github.com/sethvargo/go-retry v0.2.4/go.mod h1:1afjQuvh7s4gflMObvjLPaWgluLLyhA1wmVZ6KLpICw=
 github.com/shirou/gopsutil v3.21.11+incompatible h1:+1+c1VGhc88SSonWP6foOcLhvnKlUeu/erjjvaPEYiI=

backend/pkg/commons/db/clickhouse.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"crypto/tls"
 	"fmt"
+	"net"
 	"reflect"
 	"sync"
 	"time"
@@ -25,7 +26,15 @@ func MustInitClickhouseNative(writer *types.DatabaseConfig) ch.Conn {
 	if writer.MaxOpenConns < writer.MaxIdleConns {
 		writer.MaxIdleConns = writer.MaxOpenConns
 	}
-	log.Infof("initializing clickhouse native writer db connection to %v:%v/%v with %v/%v conn limit", writer.Host, writer.Port, writer.Name, writer.MaxIdleConns, writer.MaxOpenConns)
+	var hosts []string
+	hosts = append(hosts, net.JoinHostPort(writer.Host, writer.Port))
+	if len(writer.Failovers) > 0 {
+		for _, f := range writer.Failovers {
+			hosts = append(hosts, net.JoinHostPort(f.Host, f.Port))
+		}
+	}
+
+	log.Infof("initializing clickhouse native writer db connection to %v/%v with %v/%v conn limit", hosts, writer.Name, writer.MaxIdleConns, writer.MaxOpenConns)
 	dbWriter, err := ch.Open(&ch.Options{
 		MaxOpenConns: writer.MaxOpenConns,
 		MaxIdleConns: writer.MaxIdleConns,
@@ -34,7 +43,8 @@ func MustInitClickhouseNative(writer *types.DatabaseConfig) ch.Conn {
 		Compression: &ch.Compression{
 			Method: ch.CompressionLZ4,
 		},
-		Addr: []string{fmt.Sprintf("%s:%s", writer.Host, writer.Port)},
+		Addr:             hosts,
+		ConnOpenStrategy: ch.ConnOpenInOrder,
 		Auth: ch.Auth{
 			Username: writer.Username,
 			Password: writer.Password,

backend/pkg/commons/db/db.go

@@ -102,23 +102,31 @@ func MustInitDB(writer *types.DatabaseConfig, reader *types.DatabaseConfig, driv
 		reader.MaxIdleConns = reader.MaxOpenConns
 	}
 
-	var sslParam string
+	var extraParams string
 	if driverName == "clickhouse" {
-		sslParam = "secure=false"
+		extraParams = "secure=false"
 		if writer.SSL {
-			sslParam = "secure=true"
+			extraParams = "secure=true"
 		}
 		// debug
 		// sslParam += "&debug=true"
 	} else {
-		sslParam = "sslmode=disable"
+		extraParams = "sslmode=disable"
 		if writer.SSL {
-			sslParam = "sslmode=require"
+			extraParams = "sslmode=require"
 		}
 	}
+	var hosts string
+	hosts = net.JoinHostPort(writer.Host, writer.Port)
+	if len(writer.Failovers) > 0 {
+		for _, failover := range writer.Failovers {
+			hosts += "," + net.JoinHostPort(failover.Host, failover.Port)
+		}
+		extraParams += "&connection_open_strategy=in_order"
+	}
 
-	log.Debugf("connecting to %s database %s:%s/%s as writer with %d/%d max open/idle connections", databaseBrand, writer.Host, writer.Port, writer.Name, writer.MaxOpenConns, writer.MaxIdleConns)
-	dbConnWriter, err := sqlx.Open(driverName, fmt.Sprintf("%s://%s:%s@%s/%s?%s", databaseBrand, writer.Username, writer.Password, net.JoinHostPort(writer.Host, writer.Port), writer.Name, sslParam))
+	log.Debugf("connecting to %s database %s/%s as writer with %d/%d max open/idle connections", databaseBrand, hosts, writer.Name, writer.MaxOpenConns, writer.MaxIdleConns)
+	dbConnWriter, err := sqlx.Open(driverName, fmt.Sprintf("%s://%s:%s@%s/%s?%s", databaseBrand, writer.Username, writer.Password, hosts, writer.Name, extraParams))
 	if err != nil {
 		log.Fatal(err, "error getting Connection Writer database", 0)
 	}
@@ -134,21 +142,29 @@ func MustInitDB(writer *types.DatabaseConfig, reader *types.DatabaseConfig, driv
 	}
 
 	if driverName == "clickhouse" {
-		sslParam = "secure=false"
+		extraParams = "secure=false"
 		if writer.SSL {
-			sslParam = "secure=true"
+			extraParams = "secure=true"
 		}
 		// debug
 		// sslParam += "&debug=true"
 	} else {
-		sslParam = "sslmode=disable"
+		extraParams = "sslmode=disable"
 		if writer.SSL {
-			sslParam = "sslmode=require"
+			extraParams = "sslmode=require"
+		}
+	}
+
+	hosts = net.JoinHostPort(reader.Host, reader.Port)
+	if len(reader.Failovers) > 0 {
+		for _, failover := range reader.Failovers {
+			hosts += "," + net.JoinHostPort(failover.Host, failover.Port)
 		}
+		extraParams += "&connection_open_strategy=in_order"
 	}
 
-	log.Debugf("connecting to %s database %s:%s/%s as reader with %d/%d max open/idle connections", databaseBrand, reader.Host, reader.Port, reader.Name, reader.MaxOpenConns, reader.MaxIdleConns)
-	dbConnReader, err := sqlx.Open(driverName, fmt.Sprintf("%s://%s:%s@%s/%s?%s", databaseBrand, reader.Username, reader.Password, net.JoinHostPort(reader.Host, reader.Port), reader.Name, sslParam))
+	log.Debugf("connecting to %s database %s/%s as reader with %d/%d max open/idle connections", databaseBrand, hosts, reader.Name, reader.MaxOpenConns, reader.MaxIdleConns)
+	dbConnReader, err := sqlx.Open(driverName, fmt.Sprintf("%s://%s:%s@%s/%s?%s", databaseBrand, reader.Username, reader.Password, hosts, reader.Name, extraParams))
 	if err != nil {
 		log.Fatal(err, "error getting Connection Reader database", 0)
 	}