TheArchive/zitadel
1
0
Fork 0
mirror of https://github.com/zitadel/zitadel.git synced 2025-08-11 21:37:32 +00:00
Code Issues Packages Projects Releases Wiki Activity

refactor(database): exchange connection pool (#8325)

Browse Source 
# Which Problems Are Solved

The connection pool of go uses a high amount of database connections.

# How the Problems Are Solved

The standard lib connection pool was replaced by `pgxpool.Pool`

# Additional Changes

The `db.BeginTx`-spans are removed because they cause to much noise in
the traces.

# Additional Context

- part of https://github.com/zitadel/zitadel/issues/7639
This commit is contained in:
Silvan
2024-07-17 17:16:02 +02:00
committed by GitHub
parent d95c9508a3
commit 99c645cc60
13 changed files with 51 additions and 50 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
internal/crypto/database/database.go
View File

@@ -8,7 +8,6 @@ import (
"github.com/zitadel/zitadel/internal/crypto"
z_db "github.com/zitadel/zitadel/internal/database"
"github.com/zitadel/zitadel/internal/telemetry/tracing"
"github.com/zitadel/zitadel/internal/zerrors"
)
@@ -114,9 +113,7 @@ func (d *Database) CreateKeys(ctx context.Context, keys ...*crypto.Key) error {
if err != nil {
return zerrors.ThrowInternal(err, "", "unable to insert new keys")
}
ctx, spanBeginTx := tracing.NewNamedSpan(ctx, "db.BeginTx")
tx, err := d.client.BeginTx(ctx, nil)
spanBeginTx.EndWithError(err)
if err != nil {
return zerrors.ThrowInternal(err, "", "unable to insert new keys")
}

25
internal/database/cockroach/crdb.go
View File

@@ -1,12 +1,14 @@
package cockroach
import (
"context"
"database/sql"
"strconv"
"strings"
"time"
_ "github.com/jackc/pgx/v5/stdlib"
"github.com/jackc/pgx/v5/pgxpool"
"github.com/jackc/pgx/v5/stdlib"
"github.com/mitchellh/mapstructure"
"github.com/zitadel/logging"
@@ -70,22 +72,29 @@ func (_ *Config) Decode(configs []interface{}) (dialect.Connector, error) {
}
func (c *Config) Connect(useAdmin bool, pusherRatio, spoolerRatio float64, purpose dialect.DBPurpose) (*sql.DB, error) {
client, err := sql.Open("pgx", c.String(useAdmin, purpose.AppName()))
connConfig, err := dialect.NewConnectionConfig(c.MaxOpenConns, c.MaxIdleConns, pusherRatio, spoolerRatio, purpose)
if err != nil {
return nil, err
}
connConfig, err := dialect.NewConnectionConfig(c.MaxOpenConns, c.MaxIdleConns, spoolerRatio, pusherRatio, purpose)
config, err := pgxpool.ParseConfig(c.String(useAdmin, purpose.AppName()))
if err != nil {
return nil, err
}
config.MaxConns = int32(connConfig.MaxOpenConns)
config.MaxConnLifetime = c.MaxConnLifetime
config.MaxConnIdleTime = c.MaxConnIdleTime
pool, err := pgxpool.NewWithConfig(context.Background(), config)
if err != nil {
return nil, err
}
client.SetMaxOpenConns(int(connConfig.MaxOpenConns))
client.SetMaxIdleConns(int(connConfig.MaxIdleConns))
client.SetConnMaxLifetime(c.MaxConnLifetime)
client.SetConnMaxIdleTime(c.MaxConnIdleTime)
if err := pool.Ping(context.Background()); err != nil {
return nil, err
}
return client, nil
return stdlib.OpenDBFromPool(pool), nil
}
func (c *Config) DatabaseName() string {

5
internal/database/database.go
View File

@@ -14,7 +14,6 @@ import (
_ "github.com/zitadel/zitadel/internal/database/cockroach"
"github.com/zitadel/zitadel/internal/database/dialect"
_ "github.com/zitadel/zitadel/internal/database/postgres"
"github.com/zitadel/zitadel/internal/telemetry/tracing"
"github.com/zitadel/zitadel/internal/zerrors"
)
@@ -39,9 +38,7 @@ func (db *DB) Query(scan func(*sql.Rows) error, query string, args ...any) error
}
func (db *DB) QueryContext(ctx context.Context, scan func(rows *sql.Rows) error, query string, args ...any) (err error) {
ctx, spanBeginTx := tracing.NewNamedSpan(ctx, "db.BeginTx")
tx, err := db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true, Isolation: sql.LevelReadCommitted})
spanBeginTx.EndWithError(err)
if err != nil {
return err
}
@@ -74,9 +71,7 @@ func (db *DB) QueryRow(scan func(*sql.Row) error, query string, args ...any) (er
}
func (db *DB) QueryRowContext(ctx context.Context, scan func(row *sql.Row) error, query string, args ...any) (err error) {
ctx, spanBeginTx := tracing.NewNamedSpan(ctx, "db.BeginTx")
tx, err := db.BeginTx(ctx, &sql.TxOptions{ReadOnly: true, Isolation: sql.LevelReadCommitted})
spanBeginTx.EndWithError(err)
if err != nil {
return err
}

28
internal/database/postgres/pg.go
View File

@@ -1,12 +1,14 @@
package postgres
import (
"context"
"database/sql"
"strconv"
"strings"
"time"
_ "github.com/jackc/pgx/v5/stdlib"
"github.com/jackc/pgx/v5/pgxpool"
"github.com/jackc/pgx/v5/stdlib"
"github.com/mitchellh/mapstructure"
"github.com/zitadel/logging"
@@ -71,22 +73,32 @@ func (_ *Config) Decode(configs []interface{}) (dialect.Connector, error) {
}
func (c *Config) Connect(useAdmin bool, pusherRatio, spoolerRatio float64, purpose dialect.DBPurpose) (*sql.DB, error) {
client, err := sql.Open("pgx", c.String(useAdmin, purpose.AppName()))
connConfig, err := dialect.NewConnectionConfig(c.MaxOpenConns, c.MaxIdleConns, pusherRatio, spoolerRatio, purpose)
if err != nil {
return nil, err
}
connConfig, err := dialect.NewConnectionConfig(c.MaxOpenConns, c.MaxIdleConns, spoolerRatio, pusherRatio, purpose)
config, err := pgxpool.ParseConfig(c.String(useAdmin, purpose.AppName()))
if err != nil {
return nil, err
}
config.MaxConns = int32(connConfig.MaxOpenConns)
config.MaxConnLifetime = c.MaxConnLifetime
config.MaxConnIdleTime = c.MaxConnIdleTime
pool, err := pgxpool.NewWithConfig(
context.Background(),
config,
)
if err != nil {
return nil, err
}
client.SetMaxOpenConns(int(connConfig.MaxOpenConns))
client.SetMaxIdleConns(int(connConfig.MaxIdleConns))
client.SetConnMaxLifetime(c.MaxConnLifetime)
client.SetConnMaxIdleTime(c.MaxConnIdleTime)
if err := pool.Ping(context.Background()); err != nil {
return nil, err
}
return client, nil
return stdlib.OpenDBFromPool(pool), nil
}
func (c *Config) DatabaseName() string {

3
internal/eventstore/handler/v2/field_handler.go
View File

@@ -10,7 +10,6 @@ import (
"github.com/jackc/pgx/v5/pgconn"
"github.com/zitadel/zitadel/internal/eventstore"
"github.com/zitadel/zitadel/internal/telemetry/tracing"
)
type FieldHandler struct {
@@ -98,9 +97,7 @@ func (h *FieldHandler) processEvents(ctx context.Context, config *triggerConfig)
defer cancel()
}
ctx, spanBeginTx := tracing.NewNamedSpan(ctx, "db.BeginTx")
tx, err := h.client.BeginTx(txCtx, nil)
spanBeginTx.EndWithError(err)
if err != nil {
return false, err
}

3
internal/eventstore/handler/v2/handler.go
View File

@@ -20,7 +20,6 @@ import (
"github.com/zitadel/zitadel/internal/migration"
"github.com/zitadel/zitadel/internal/repository/instance"
"github.com/zitadel/zitadel/internal/repository/pseudo"
"github.com/zitadel/zitadel/internal/telemetry/tracing"
)
type EventStore interface {
@@ -477,9 +476,7 @@ func (h *Handler) processEvents(ctx context.Context, config *triggerConfig) (add
defer cancel()
}
ctx, spanBeginTx := tracing.NewNamedSpan(ctx, "db.BeginTx")
tx, err := h.client.BeginTx(txCtx, nil)
spanBeginTx.EndWithError(err)
if err != nil {
return false, err
}

3
internal/eventstore/v3/push.go
View File

@@ -14,14 +14,11 @@ import (
"github.com/zitadel/logging"
"github.com/zitadel/zitadel/internal/eventstore"
"github.com/zitadel/zitadel/internal/telemetry/tracing"
"github.com/zitadel/zitadel/internal/zerrors"
)
func (es *Eventstore) Push(ctx context.Context, commands ...eventstore.Command) (events []eventstore.Event, err error) {
ctx, spanBeginTx := tracing.NewNamedSpan(ctx, "db.BeginTx")
tx, err := es.client.BeginTx(ctx, nil)
spanBeginTx.EndWithError(err)
if err != nil {
return nil, err
}

3
internal/integration/config/postgres.yaml
View File

@@ -3,8 +3,7 @@ Database:
Host: localhost
Port: 5432
Database: zitadel
EventPushConnRatio: 0.2
MaxOpenConns: 40
MaxOpenConns: 15
MaxIdleConns: 10
User:
Username: zitadel

4
internal/query/current_state.go
View File

@@ -90,9 +90,7 @@ func (q *Queries) latestState(ctx context.Context, projections ...table) (state
}
func (q *Queries) ClearCurrentSequence(ctx context.Context, projectionName string) (err error) {
ctx, spanBeginTx := tracing.NewNamedSpan(ctx, "db.BeginTx")
tx, err := q.client.BeginTx(ctx, nil)
spanBeginTx.EndWithError(err)
if err != nil {
return zerrors.ThrowInternal(err, "QUERY-9iOpr", "Errors.RemoveFailed")
}
@@ -205,7 +203,7 @@ func reset(ctx context.Context, tx *sql.Tx, tables []string, projectionName stri
if err != nil {
return zerrors.ThrowInternal(err, "QUERY-Ff3tw", "Errors.RemoveFailed")
}
_, err = tx.Exec(update, args...)
_, err = tx.ExecContext(ctx, update, args...)
if err != nil {
return zerrors.ThrowInternal(err, "QUERY-NFiws", "Errors.RemoveFailed")
}