Merging upstream version 1.2.0.

Signed-off-by: Daniel Baumann <daniel.baumann@progress-linux.org>

1 files changed, 58 insertions, 17 deletions

diff --git a/pkg/config/database.go b/pkg/config/database.go
index b42ff8e..0895d26 100644
--- a/pkg/config/database.go
+++ b/pkg/config/database.go
@@ -1,25 +1,25 @@
 package config
 
 import (
+	"context"
+	"database/sql"
+	"database/sql/driver"
 	"fmt"
 	"github.com/go-sql-driver/mysql"
-	"github.com/icinga/icingadb/pkg/driver"
 	"github.com/icinga/icingadb/pkg/icingadb"
 	"github.com/icinga/icingadb/pkg/logging"
 	"github.com/icinga/icingadb/pkg/utils"
 	"github.com/jmoiron/sqlx"
 	"github.com/jmoiron/sqlx/reflectx"
+	"github.com/lib/pq"
 	"github.com/pkg/errors"
 	"net"
 	"net/url"
 	"strconv"
 	"strings"
-	"sync"
 	"time"
 )
 
-var registerDriverOnce sync.Once
-
 // Database defines database client configuration.
 type Database struct {
 	Type       string           `yaml:"type" default:"mysql"`
@@ -35,17 +35,14 @@ type Database struct {
 // Open prepares the DSN string and driver configuration,
 // calls sqlx.Open, but returns *icingadb.DB.
 func (d *Database) Open(logger *logging.Logger) (*icingadb.DB, error) {
-	registerDriverOnce.Do(func() {
-		driver.Register(logger)
-	})
-
-	var dsn string
+	var db *sqlx.DB
 	switch d.Type {
 	case "mysql":
 		config := mysql.NewConfig()
 
 		config.User = d.User
 		config.Passwd = d.Password
+		config.Logger = icingadb.MysqlFuncLogger(logger.Debug)
 
 		if d.isUnixAddr() {
 			config.Net = "unix"
@@ -61,7 +58,7 @@ func (d *Database) Open(logger *logging.Logger) (*icingadb.DB, error) {
 
 		config.DBName = d.Database
 		config.Timeout = time.Minute
-		config.Params = map[string]string{"sql_mode": "ANSI_QUOTES"}
+		config.Params = map[string]string{"sql_mode": "'TRADITIONAL,ANSI_QUOTES'"}
 
 		tlsConfig, err := d.TlsOptions.MakeConfig(d.Host)
 		if err != nil {
@@ -75,7 +72,17 @@ func (d *Database) Open(logger *logging.Logger) (*icingadb.DB, error) {
 			}
 		}
 
-		dsn = config.FormatDSN()
+		c, err := mysql.NewConnector(config)
+		if err != nil {
+			return nil, errors.Wrap(err, "can't open mysql database")
+		}
+
+		wsrepSyncWait := int64(d.Options.WsrepSyncWait)
+		setWsrepSyncWait := func(ctx context.Context, conn driver.Conn) error {
+			return setGaleraOpts(ctx, conn, wsrepSyncWait)
+		}
+
+		db = sqlx.NewDb(sql.OpenDB(icingadb.NewConnector(c, logger, setWsrepSyncWait)), icingadb.MySQL)
 	case "pgsql":
 		uri := &url.URL{
 			Scheme: "postgres",
@@ -123,16 +130,17 @@ func (d *Database) Open(logger *logging.Logger) (*icingadb.DB, error) {
 		}
 
 		uri.RawQuery = query.Encode()
-		dsn = uri.String()
+
+		connector, err := pq.NewConnector(uri.String())
+		if err != nil {
+			return nil, errors.Wrap(err, "can't open pgsql database")
+		}
+
+		db = sqlx.NewDb(sql.OpenDB(icingadb.NewConnector(connector, logger, nil)), icingadb.PostgreSQL)
 	default:
 		return nil, unknownDbType(d.Type)
 	}
 
-	db, err := sqlx.Open("icingadb-"+d.Type, dsn)
-	if err != nil {
-		return nil, errors.Wrap(err, "can't open database")
-	}
-
 	db.SetMaxIdleConns(d.Options.MaxConnections / 3)
 	db.SetMaxOpenConns(d.Options.MaxConnections)
 
@@ -173,3 +181,36 @@ func (d *Database) isUnixAddr() bool {
 func unknownDbType(t string) error {
 	return errors.Errorf(`unknown database type %q, must be one of: "mysql", "pgsql"`, t)
 }
+
+// setGaleraOpts sets the "wsrep_sync_wait" variable for each session ensures that causality checks are performed
+// before execution and that each statement is executed on a fully synchronized node. Doing so prevents foreign key
+// violation when inserting into dependent tables on different MariaDB/MySQL nodes. When using MySQL single nodes,
+// the "SET SESSION" command will fail with "Unknown system variable (1193)" and will therefore be silently dropped.
+//
+// https://mariadb.com/kb/en/galera-cluster-system-variables/#wsrep_sync_wait
+func setGaleraOpts(ctx context.Context, conn driver.Conn, wsrepSyncWait int64) error {
+	const galeraOpts = "SET SESSION wsrep_sync_wait=?"
+
+	stmt, err := conn.(driver.ConnPrepareContext).PrepareContext(ctx, galeraOpts)
+	if err != nil {
+		if errors.Is(err, &mysql.MySQLError{Number: 1193}) { // Unknown system variable
+			return nil
+		}
+
+		return errors.Wrap(err, "cannot prepare "+galeraOpts)
+	}
+	// This is just for an unexpected exit and any returned error can safely be ignored and in case
+	// of the normal function exit, the stmt is closed manually, and its error is handled gracefully.
+	defer func() { _ = stmt.Close() }()
+
+	_, err = stmt.(driver.StmtExecContext).ExecContext(ctx, []driver.NamedValue{{Value: wsrepSyncWait}})
+	if err != nil {
+		return errors.Wrap(err, "cannot execute "+galeraOpts)
+	}
+
+	if err = stmt.Close(); err != nil {
+		return errors.Wrap(err, "cannot close prepared statement "+galeraOpts)
+	}
+
+	return nil
+}