Set db connection read and write timeout (#539)

Fix - when the database goes down, or the network dies, the open connections will hang
 1. add library pq-timeouts and set read/write timeouts from the configuration
 2. reduce notification clean interval to 15 minutes
 3. fix tests - In current test implementation, the configuration for DB sslmode is passed as part of the DB URI string 
     passed by settings.URI - "postgres://postgres:postgres@localhost:5432/postgres?sslmode=disable".  
     Because of it, I can't just simply concatenate the read/write new parameters like done in previous commits. I need to 
     take into account that the URI string might contain already URL parameters.

Gopkg.toml

@@ -45,6 +45,10 @@
   name = "github.com/lib/pq"
   branch = "master"
 
+[[constraint]]
+  name = "github.com/Kount/pq-timeouts"
+  version = "=1.0.0"
+
 [[constraint]]
   name = "github.com/gofrs/uuid"
   version = "=3.1.0"

pkg/sm/sm.go

@@ -54,6 +54,7 @@ import (
 	"github.com/Peripli/service-manager/storage"
 	"github.com/Peripli/service-manager/storage/postgres"
 
+	_ "github.com/Kount/pq-timeouts"
 	"github.com/Peripli/service-manager/api/filters"
 	"github.com/Peripli/service-manager/pkg/web"
 	osbc "github.com/kubernetes-sigs/go-open-service-broker-client/v2"

storage/interfaces.go

@@ -64,6 +64,8 @@ type Settings struct {
 	SkipSSLValidation  bool                  `mapstructure:"skip_ssl_validation" description:"whether to skip ssl verification when connecting to the storage"`
 	MaxIdleConnections int                   `mapstructure:"max_idle_connections" description:"sets the maximum number of connections in the idle connection pool"`
 	MaxOpenConnections int                   `mapstructure:"max_open_connections" description:"sets the maximum number of open connections to the database"`
+	ReadTimeout        int                   `mapstructure:"read_timeout" description:"sets the limit for reading in milliseconds"`
+	WriteTimeout       int                   `mapstructure:"write_timeout" description:"sets the limit for writing in milliseconds"`
 	Notification       *NotificationSettings `mapstructure:"notification"`
 	IntegrityProcessor security.IntegrityProcessor
 }
@@ -77,6 +79,8 @@ func DefaultSettings() *Settings {
 		SkipSSLValidation:  false,
 		MaxIdleConnections: 5,
 		MaxOpenConnections: 30,
+		ReadTimeout:        900000, //15 minutes
+		WriteTimeout:       900000, //15 minutes
 		Notification:       DefaultNotificationSettings(),
 		IntegrityProcessor: &security.HashingIntegrityProcessor{
 			HashingFunc: func(data []byte) []byte {
@@ -119,7 +123,7 @@ func DefaultNotificationSettings() *NotificationSettings {
 		QueuesSize:           100,
 		MinReconnectInterval: time.Millisecond * 200,
 		MaxReconnectInterval: time.Second * 20,
-		CleanInterval:        time.Hour,
+		CleanInterval:        time.Minute * 15,
 		KeepFor:              time.Hour * 12,
 	}
 }

storage/postgres/storage.go

@@ -20,6 +20,8 @@ import (
 	"context"
 	"database/sql"
 	"fmt"
+	"net/url"
+	"strconv"
 	"strings"
 	"sync"
 	"time"
@@ -39,7 +41,7 @@ import (
 )
 
 const (
-	postgresDriverName  = "postgres"
+	postgresDriverName  = "pq-timeouts"
 	foreignKeyViolation = "foreign_key_violation"
 )
 
@@ -72,11 +74,24 @@ func (ps *Storage) Open(settings *storage.Settings) error {
 	ps.mutex.Lock()
 	defer ps.mutex.Unlock()
 	if ps.db == nil {
-		sslModeParam := ""
+		parsedUrl, err := url.Parse(settings.URI)
+		if err != nil {
+			return fmt.Errorf("could not parse PostgreSQL URI: %s", err)
+		}
+
+		parsedQuery, err := url.ParseQuery(parsedUrl.RawQuery)
+		if err != nil {
+			return fmt.Errorf("could not parse PostgreSQL URL query: %s", err)
+		}
+
+		parsedQuery.Set("read_timeout", strconv.Itoa(settings.ReadTimeout))
+		parsedQuery.Set("write_timeout", strconv.Itoa(settings.WriteTimeout))
 		if settings.SkipSSLValidation {
-			sslModeParam = "?sslmode=disable"
+			parsedQuery.Add("sslmode", "disable")
 		}
-		db, err := ps.ConnectFunc(postgresDriverName, settings.URI+sslModeParam)
+		parsedUrl.RawQuery = parsedQuery.Encode()
+
+		db, err := ps.ConnectFunc(postgresDriverName, parsedUrl.String())
 		if err != nil {
 			return fmt.Errorf("could not connect to PostgreSQL: %s", err)
 		}