Merge pull request #80 from practo/keep-alive

Keep alive support for redshift connection

Author: alok87

redshiftsink/Makefile

@@ -25,7 +25,7 @@ BASEIMAGE ?= gcr.io/distroless/static
 
 TAG := $(VERSION)
 
-BUILD_IMAGE ?= golang:1.15.1-alpine
+BUILD_IMAGE ?= golang:1.15.2-alpine
 
 # If you want to build all binaries, see the 'all-build' rule.
 # If you want to build all containers, see the 'all-container' rule.

redshiftsink/go.mod

@@ -6,10 +6,10 @@ require (
 	github.com/Shopify/sarama v1.26.5-0.20200810201100-25aedae6c37b
 	github.com/aws/aws-sdk-go v1.34.1
 	github.com/hashicorp/go-multierror v1.0.0
-	github.com/lib/pq v1.8.0
 	github.com/linkedin/goavro/v2 v2.9.8
 	github.com/practo/gobatch v0.0.0-20200822085922-4905d08d9f40
 	github.com/practo/klog/v2 v2.2.1
+	github.com/practo/pq v0.0.0-20200930024154-af3ceb106a20
 	github.com/prometheus/client_golang v1.7.0
 	github.com/riferrei/srclient v0.0.0-20200820123217-6a5d2a4bf9d1
 	github.com/spf13/cobra v1.0.0

redshiftsink/go.sum

@@ -169,8 +169,6 @@ github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
 github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
 github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
 github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
-github.com/lib/pq v1.8.0 h1:9xohqzkUwzR4Ga4ivdTcawVS89YSDVxXMa3xJX3cGzg=
-github.com/lib/pq v1.8.0/go.mod h1:AlVN5x4E4T544tWzH6hKfbfQvm3HdbOxrmggDNAPY9o=
 github.com/linkedin/goavro/v2 v2.9.7/go.mod h1:UgQUb2N/pmueQYH9bfqFioWxzYCZXSfF8Jw03O5sjqA=
 github.com/linkedin/goavro/v2 v2.9.8 h1:jN50elxBsGBDGVDEKqUlDuU1cFwJ11K/yrJCBMe/7Wg=
 github.com/linkedin/goavro/v2 v2.9.8/go.mod h1:UgQUb2N/pmueQYH9bfqFioWxzYCZXSfF8Jw03O5sjqA=
@@ -214,6 +212,8 @@ github.com/practo/gobatch v0.0.0-20200822085922-4905d08d9f40 h1:ukO+fKWBM3U9LYZh
 github.com/practo/gobatch v0.0.0-20200822085922-4905d08d9f40/go.mod h1:OR//ozfD9hhQvQ1exBwHezn3VtEV1BjaLWj1LP+gAAk=
 github.com/practo/klog/v2 v2.2.1 h1:yV/SavG73KHSmCVwtDap1RGe+2jd86e1eugT+Mv2/FA=
 github.com/practo/klog/v2 v2.2.1/go.mod h1:WMkpfPwTxLgSLookpI4UUv2zL7tMltKACK/FHB1zLV0=
+github.com/practo/pq v0.0.0-20200930024154-af3ceb106a20 h1:05vUPBE8rE/C036Jb/Rzt4699sv+GzZSxAWVlw0wblk=
+github.com/practo/pq v0.0.0-20200930024154-af3ceb106a20/go.mod h1:C7JAQ0g3vCXYh7S7oWN8aTirSBqT8Xi4bUbGi1jP9Ak=
 github.com/prometheus/client_golang v0.9.1/go.mod h1:7SWBe2y4D6OKWSNQJUaRYU/AaXPKyh/dDVn+NZz0KFw=
 github.com/prometheus/client_golang v0.9.3/go.mod h1:/TN21ttK/J9q6uSwhBd54HahCDft0ttaMvbicHlPoso=
 github.com/prometheus/client_golang v1.0.0/go.mod h1:db9x61etRT2tGnBNRi70OPL5FsnadC4Ky3P0J6CfImo=

redshiftsink/pkg/redshift/redshift.go

@@ -11,7 +11,8 @@ import (
 	// Use our own version of the postgres library so we get keep-alive support.
 	// See https://github.com/Clever/pq/pull/1
 	// TODO: https://github.com/Clever/s3-to-redshift/issues/163
-	_ "github.com/lib/pq"
+	// TCP keep alive support pending: https://github.com/lib/pq/issues/360
+	_ "github.com/practo/pq"
 	"strings"
 )
 
@@ -68,6 +69,9 @@ type dbExecCloser interface {
 	QueryContext(
 		c context.Context, q string, a ...interface{}) (*sql.Rows, error)
 	QueryRowContext(c context.Context, q string, a ...interface{}) *sql.Row
+	// tcp max conn time
+	// https://github.com/lib/pq/issues/360#issuecomment-565121408
+	// SetConnMaxLifetime(d time.Duration)
 }
 
 // Redshift wraps a dbExecCloser and can be used to perform
@@ -128,7 +132,7 @@ type ColInfo struct {
 
 func NewRedshift(ctx context.Context, conf RedshiftConfig) (*Redshift, error) {
 	source := fmt.Sprintf(
-		"host=%s port=%s dbname=%s connect_timeout=%d",
+		"host=%s port=%s dbname=%s keepalive=1 connect_timeout=%d",
 		conf.Host, conf.Port, conf.Database, conf.Timeout,
 	)
 	// TODO: make ssl configurable
@@ -141,7 +145,13 @@ func NewRedshift(ctx context.Context, conf RedshiftConfig) (*Redshift, error) {
 	if err := sqldb.Ping(); err != nil {
 		return nil, err
 	}
-	return &Redshift{sqldb, ctx, conf}, nil
+	r := &Redshift{sqldb, ctx, conf}
+
+	// TODO: not using this
+	// klog.Info("Setting Redshift ConnMaxLifetime=-1 (keep alive)")
+	// r.SetConnMaxLifetime(1200 * time.Second)
+
+	return r, nil
 }
 
 // Begin wraps a new transaction in the databases context