kuma

1
package postgres
2

3
import (
4
	"fmt"
5
	"strings"
6
	"time"
7

8
	"github.com/pkg/errors"
9

10
	"github.com/kumahq/kuma/pkg/config"
11
	config_types "github.com/kumahq/kuma/pkg/config/types"
12
	"github.com/kumahq/kuma/pkg/core"
13
)
14

15
const (
16
	DriverNamePgx             = "pgx"
17
	DriverNamePq              = "postgres"
18
	DefaultMaxIdleConnections = 50
19
	DefaultMinOpenConnections = 0
20
)
21

22
var (
23
	_      config.Config = &PostgresStoreConfig{}
24
	logger               = core.Log.WithName("postgres-config")
25
)
26

27
var (
28
	DefaultMinReconnectInterval        = config_types.Duration{Duration: 10 * time.Second}
29
	DefaultMaxReconnectInterval        = config_types.Duration{Duration: 60 * time.Second}
30
	DefaultMaxConnectionLifetime       = config_types.Duration{Duration: time.Hour}
31
	DefaultMaxConnectionLifetimeJitter = config_types.Duration{Duration: 1 * time.Minute}
32
	DefaultHealthCheckInterval         = config_types.Duration{Duration: 30 * time.Second}
33
	// some of the above settings taken from pgx https://github.com/jackc/pgx/blob/ca022267dbbfe7a8ba7070557352a5cd08f6cb37/pgxpool/pool.go#L18-L22
34
)
35

36
// PostgresStoreConfig defines Postgres store configuration
37
type PostgresStoreConfig struct {
38
	config.BaseConfig
39

40
	// Host of the Postgres DB
41
	Host string `json:"host" envconfig:"kuma_store_postgres_host"`
42
	// Port of the Postgres DB
43
	Port int `json:"port" envconfig:"kuma_store_postgres_port"`
44
	// User of the Postgres DB
45
	User string `json:"user" envconfig:"kuma_store_postgres_user"`
46
	// Password of the Postgres DB
47
	Password string `json:"password" envconfig:"kuma_store_postgres_password"`
48
	// Database name of the Postgres DB
49
	DbName string `json:"dbName" envconfig:"kuma_store_postgres_db_name"`
50
	// Driver to use, one of: pgx, postgres
51
	DriverName string `json:"driverName" envconfig:"kuma_store_postgres_driver_name"`
52
	// Connection Timeout to the DB in seconds
53
	ConnectionTimeout int `json:"connectionTimeout" envconfig:"kuma_store_postgres_connection_timeout"`
54
	// MaxConnectionLifetime (applied only when driverName=pgx) is the duration since creation after which a connection will be automatically closed
55
	MaxConnectionLifetime config_types.Duration `json:"maxConnectionLifetime" envconfig:"kuma_store_postgres_max_connection_lifetime"`
56
	// MaxConnectionLifetimeJitter (applied only when driverName=pgx) is the duration after MaxConnectionLifetime to randomly decide to close a connection.
57
	// This helps prevent all connections from being closed at the exact same time, starving the pool.
58
	MaxConnectionLifetimeJitter config_types.Duration `json:"maxConnectionLifetimeJitter" envconfig:"kuma_store_postgres_max_connection_lifetime_jitter"`
59
	// HealthCheckInterval (applied only when driverName=pgx) is the duration between checks of the health of idle connections.
60
	HealthCheckInterval config_types.Duration `json:"healthCheckInterval" envconfig:"kuma_store_postgres_health_check_interval"`
61
	// MinOpenConnections (applied only when driverName=pgx) is the minimum number of open connections to the database
62
	MinOpenConnections int `json:"minOpenConnections" envconfig:"kuma_store_postgres_min_open_connections"`
63
	// MaxOpenConnections is the maximum number of open connections to the database
64
	// `0` value means number of open connections is unlimited
65
	MaxOpenConnections int `json:"maxOpenConnections" envconfig:"kuma_store_postgres_max_open_connections"`
66
	// MaxIdleConnections (applied only when driverName=postgres) is the maximum number of connections in the idle connection pool
67
	// <0 value means no idle connections and 0 means default max idle connections
68
	// Deprecated: it's only used when driverName=postgres (lib/pq) which is deprecated, use driverName=pgx instead.
69
	MaxIdleConnections int `json:"maxIdleConnections" envconfig:"kuma_store_postgres_max_idle_connections"`
70
	// TLS settings
71
	TLS TLSPostgresStoreConfig `json:"tls"`
72
	// MinReconnectInterval (applied only when driverName=postgres) controls the duration to wait before trying to
73
	// re-establish the database connection after connection loss. After each
74
	// consecutive failure this interval is doubled, until MaxReconnectInterval
75
	// is reached. Successfully completing the connection establishment procedure
76
	// resets the interval back to MinReconnectInterval.
77
	// Deprecated: it's only used when driverName=postgres (lib/pq) which is deprecated, use driverName=pgx instead.
78
	MinReconnectInterval config_types.Duration `json:"minReconnectInterval" envconfig:"kuma_store_postgres_min_reconnect_interval"`
79
	// MaxReconnectInterval (applied only when driverName=postgres) controls the maximum possible duration to wait before trying
80
	// to re-establish the database connection after connection loss.
81
	// Deprecated: it's only used when driverName=postgres (lib/pq) which is deprecated, use driverName=pgx instead.
82
	MaxReconnectInterval config_types.Duration `json:"maxReconnectInterval" envconfig:"kuma_store_postgres_max_reconnect_interval"`
83
	// MaxListQueryElements defines maximum number of changed elements before requesting full list of elements from the store.
84
	MaxListQueryElements uint32 `json:"maxListQueryElements" envconfig:"kuma_store_postgres_max_list_query_elements"`
85
	// ReadReplica is a setting for a DB replica used only for read queries
86
	ReadReplica ReadReplica `json:"readReplica"`
87
}
88

89
type ReadReplica struct {
90
	// Host of the Postgres DB read replica. If not set, read replica is not used.
91
	Host string `json:"host" envconfig:"kuma_store_postgres_read_replica_host"`
92
	// Port of the Postgres DB read replica
93
	Port uint `json:"port" envconfig:"kuma_store_postgres_read_replica_port"`
94
	// Ratio in [0-100] range. How many SELECT queries (out of 100) will use read replica.
95
	Ratio uint `json:"ratio" envconfig:"kuma_store_postgres_read_replica_ratio"`
96
}
97

98
func (cfg ReadReplica) Validate() error {
99
	if cfg.Ratio > 100 {
100
		return errors.New(".Ratio out of [0-100] range")
101
	}
102
	return nil
103
}
104

105
func (cfg PostgresStoreConfig) ConnectionString() (string, error) {
106
	mode, err := cfg.TLS.Mode.postgresMode()
107
	if err != nil {
108
		return "", err
109
	}
110
	escape := func(value string) string { return strings.ReplaceAll(strings.ReplaceAll(value, `\`, `\\`), `'`, `\'`) }
111
	intVariable := func(name string, value int) string {
112
		return fmt.Sprintf("%s=%d", name, value)
113
	}
114
	variable := func(name, value string) string {
115
		return fmt.Sprintf("%s=%s", name, value)
116
	}
117
	quotedVariable := func(name, value string) string {
118
		return fmt.Sprintf("%s='%s'", name, escape(value))
119
	}
120
	variables := []string{
121
		quotedVariable("host", cfg.Host),
122
		intVariable("port", cfg.Port),
123
		quotedVariable("user", cfg.User),
124
		quotedVariable("password", cfg.Password),
125
		quotedVariable("dbname", cfg.DbName),
126
		intVariable("connect_timeout", cfg.ConnectionTimeout),
127
		variable("sslmode", mode),
128
		quotedVariable("sslcert", cfg.TLS.CertPath),
129
		quotedVariable("sslkey", cfg.TLS.KeyPath),
130
		quotedVariable("sslrootcert", cfg.TLS.CAPath),
131
	}
132
	if cfg.TLS.DisableSSLSNI {
133
		variables = append(variables, "sslsni=0")
134
	}
135
	return strings.Join(variables, " "), nil
136
}
137

138
// TLSMode modes available here https://godoc.org/github.com/lib/pq
139
type TLSMode string
140

141
const (
142
	Disable TLSMode = "disable"
143
	// VerifyNone represents Always TLS (skip verification)
144
	VerifyNone TLSMode = "verifyNone"
145
	// VerifyCa represents Always TLS (verify that the certificate presented by the server was signed by a trusted CA)
146
	VerifyCa TLSMode = "verifyCa"
147
	// VerifyFull represents Always TLS (verify that the certification presented by the server was signed by a trusted CA and the server host name matches the one in the certificate)
148
	VerifyFull TLSMode = "verifyFull"
149
)
150

151
func (mode TLSMode) postgresMode() (string, error) {
152
	switch mode {
153
	case Disable:
154
		return "disable", nil
155
	case VerifyNone:
156
		return "require", nil
157
	case VerifyCa:
158
		return "verify-ca", nil
159
	case VerifyFull:
160
		return "verify-full", nil
161
	default:
162
		return "", errors.Errorf("could not translate mode %q to postgres mode", mode)
163
	}
164
}
165

166
type TLSPostgresStoreConfig struct {
167
	config.BaseConfig
168

169
	// Mode of TLS connection. Available values (disable, verifyNone, verifyCa, verifyFull)
170
	Mode TLSMode `json:"mode" envconfig:"kuma_store_postgres_tls_mode"`
171
	// Path to TLS Certificate of the client. Required when server has METHOD=cert
172
	CertPath string `json:"certPath" envconfig:"kuma_store_postgres_tls_cert_path"`
173
	// Path to TLS Key of the client. Required when server has METHOD=cert
174
	KeyPath string `json:"keyPath" envconfig:"kuma_store_postgres_tls_key_path"`
175
	// Path to the root certificate. Used in verifyCa and verifyFull modes.
176
	CAPath string `json:"caPath" envconfig:"kuma_store_postgres_tls_ca_path"`
177
	// Whether to disable SNI the postgres `sslsni` option.
178
	DisableSSLSNI bool `json:"disableSSLSNI" envconfig:"kuma_store_postgres_tls_disable_sslsni"`
179
}
180

181
func (s TLSPostgresStoreConfig) Validate() error {
182
	switch s.Mode {
183
	case VerifyFull:
184
		fallthrough
185
	case VerifyCa:
186
		if s.CAPath == "" {
187
			return errors.New("CAPath cannot be empty")
188
		}
189
	case VerifyNone:
190
	case Disable:
191
	default:
192
		return errors.Errorf("invalid mode: %s", s.Mode)
193
	}
194
	if s.KeyPath == "" && s.CertPath != "" {
195
		return errors.New("KeyPath cannot be empty when CertPath is provided")
196
	}
197
	if s.CertPath == "" && s.KeyPath != "" {
198
		return errors.New("CertPath cannot be empty when KeyPath is provided")
199
	}
200
	return nil
201
}
202

203
func (p *PostgresStoreConfig) Sanitize() {
204
	p.Password = config.SanitizedValue
205
}
206

207
func (p *PostgresStoreConfig) Validate() error {
208
	if len(p.Host) < 1 {
209
		return errors.New("Host should not be empty")
210
	}
211
	if p.Port < 0 {
212
		return errors.New("Port cannot be negative")
213
	}
214
	if len(p.User) < 1 {
215
		return errors.New("User should not be empty")
216
	}
217
	if len(p.Password) < 1 {
218
		return errors.New("Password should not be empty")
219
	}
220
	if len(p.DbName) < 1 {
221
		return errors.New("DbName should not be empty")
222
	}
223
	if err := p.TLS.Validate(); err != nil {
224
		return errors.Wrap(err, "TLS validation failed")
225
	}
226
	if p.MinReconnectInterval.Duration >= p.MaxReconnectInterval.Duration {
227
		return errors.New("MinReconnectInterval should be less than MaxReconnectInterval")
228
	}
229
	if p.MinOpenConnections < 0 {
230
		return errors.New("MinOpenConnections should be greater than 0")
231
	}
232
	if p.MinOpenConnections > p.MaxOpenConnections {
233
		return errors.New("MinOpenConnections should be less than MaxOpenConnections")
234
	}
235
	if p.MaxConnectionLifetime.Duration < 0 {
236
		return errors.New("MaxConnectionLifetime should be greater than 0")
237
	}
238
	if p.MaxConnectionLifetimeJitter.Duration < 0 {
239
		return errors.New("MaxConnectionLifetimeJitter should be greater than 0")
240
	}
241
	if p.MaxConnectionLifetimeJitter.Duration > p.MaxConnectionLifetime.Duration {
242
		return errors.New("MaxConnectionLifetimeJitter should be less than MaxConnectionLifetime")
243
	}
244
	if p.HealthCheckInterval.Duration < 0 {
245
		return errors.New("HealthCheckInterval should be greater than 0")
246
	}
247
	if err := p.ReadReplica.Validate(); err != nil {
248
		return errors.Wrapf(err, "ReadReplica validation failed")
249
	}
250
	warning := "[WARNING] "
251
	switch p.DriverName {
252
	case DriverNamePgx:
253
		pqAlternativeMessage := "does not have an equivalent for pgx driver. " +
254
			"If you need this setting consider using lib/pq as a postgres driver by setting driverName='postgres' or " +
255
			"setting KUMA_STORE_POSTGRES_DRIVER_NAME='postgres' env variable."
256
		if p.MaxIdleConnections != DefaultMaxIdleConnections {
257
			logger.Info(warning + "MaxIdleConnections " + pqAlternativeMessage)
258
		}
259
		if p.MinReconnectInterval != DefaultMinReconnectInterval {
260
			logger.Info(warning + "MinReconnectInterval " + pqAlternativeMessage)
261
		}
262
		if p.MaxReconnectInterval != DefaultMaxReconnectInterval {
263
			logger.Info(warning + "MaxReconnectInterval " + pqAlternativeMessage)
264
		}
265
	case DriverNamePq:
266
		logger.Info(warning + "DriverName='postgres' (pq) is deprecated. Please use DriverName='pgx' instead.")
267
		pgxAlternativeMessage := "does not have an equivalent for pq driver. " +
268
			"If you need this setting consider using pgx as a postgres driver by setting driverName='pgx' or " +
269
			"setting KUMA_STORE_POSTGRES_DRIVER_NAME='pgx' env variable."
270
		if p.MinOpenConnections != DefaultMinOpenConnections {
271
			logger.Info(warning + "MinOpenConnections " + pgxAlternativeMessage)
272
		}
273
		if p.MaxConnectionLifetime != DefaultMaxConnectionLifetime {
274
			logger.Info(warning + "MaxConnectionLifetime " + pgxAlternativeMessage)
275
		}
276
		if p.MaxConnectionLifetimeJitter != DefaultMaxConnectionLifetimeJitter {
277
			logger.Info(warning + "MaxConnectionLifetimeJitter " + pgxAlternativeMessage)
278
		}
279
		if p.HealthCheckInterval != DefaultHealthCheckInterval {
280
			logger.Info(warning + "HealthCheckInterval " + pgxAlternativeMessage)
281
		}
282
	}
283

284
	return nil
285
}
286

287
func DefaultPostgresStoreConfig() *PostgresStoreConfig {
288
	return &PostgresStoreConfig{
289
		Host:                        "127.0.0.1",
290
		Port:                        15432,
291
		User:                        "kuma",
292
		Password:                    "kuma",
293
		DbName:                      "kuma",
294
		DriverName:                  DriverNamePgx,
295
		ConnectionTimeout:           5,
296
		MaxOpenConnections:          50,                        // 0 for unlimited
297
		MaxIdleConnections:          DefaultMaxIdleConnections, // 0 for unlimited
298
		TLS:                         DefaultTLSPostgresStoreConfig(),
299
		MinReconnectInterval:        DefaultMinReconnectInterval,
300
		MaxReconnectInterval:        DefaultMaxReconnectInterval,
301
		MinOpenConnections:          DefaultMinOpenConnections,
302
		MaxConnectionLifetime:       DefaultMaxConnectionLifetime,
303
		MaxConnectionLifetimeJitter: DefaultMaxConnectionLifetimeJitter,
304
		HealthCheckInterval:         DefaultHealthCheckInterval,
305
		MaxListQueryElements:        0,
306
		ReadReplica: ReadReplica{
307
			Port:  5432,
308
			Ratio: 100,
309
		},
310
	}
311
}
312

313
var _ config.Config = &TLSPostgresStoreConfig{}
314

315
func DefaultTLSPostgresStoreConfig() TLSPostgresStoreConfig {
316
	return TLSPostgresStoreConfig{
317
		Mode:     Disable,
318
		CertPath: "",
319
		KeyPath:  "",
320
		CAPath:   "",
321
	}
322
}
323