mirror of
https://github.com/zitadel/zitadel.git
synced 2024-12-14 11:58:02 +00:00
9c3e5e467b
# Which Problems Are Solved Queries currently execute 3 statements, begin, query, commit # How the Problems Are Solved remove transaction handling from query methods in database package # Additional Changes - Bump versions of `core_grpc_dependencies`-receipt in Makefile # Additional info During load tests we saw a lot of idle transactions of `zitadel_queries` application name which is the connection pool used to query data in zitadel. Executed query: `select query_start - xact_start, pid, application_name, backend_start, xact_start, query_start, state_change, wait_event_type, wait_event,substring(query, 1, 200) query from pg_stat_activity where datname = 'zitadel' and state <> 'idle';` Mostly the last query executed was `begin isolation level read committed read only`. example: ``` ?column? | pid | application_name | backend_start | xact_start | query_start | state_change | wait_event_type | wait_event | query -----------------+-------+----------------------------+-------------------------------+-------------------------------+-------------------------------+-------------------------------+-----------------+--------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- 00:00:00 | 33030 | zitadel_queries | 2024-10-16 16:25:53.906036+00 | 2024-10-16 16:30:19.191661+00 | 2024-10-16 16:30:19.191661+00 | 2024-10-16 16:30:19.19169+00 | Client | ClientRead | begin isolation level read committed read only 00:00:00 | 33035 | zitadel_queries | 2024-10-16 16:25:53.909629+00 | 2024-10-16 16:30:19.19179+00 | 2024-10-16 16:30:19.19179+00 | 2024-10-16 16:30:19.191805+00 | Client | ClientRead | begin isolation level read committed read only 00:00:00.00412 | 33028 | zitadel_queries | 2024-10-16 16:25:53.904247+00 | 2024-10-16 16:30:19.187734+00 | 2024-10-16 16:30:19.191854+00 | 2024-10-16 16:30:19.191964+00 | Client | ClientRead | SELECT created_at, event_type, "sequence", "position", payload, creator, "owner", instance_id, aggregate_type, aggregate_id, revision FROM eventstore.events2 WHERE instance_id = $1 AND aggregate_type 00:00:00.084662 | 33134 | zitadel_es_pusher | 2024-10-16 16:29:54.979692+00 | 2024-10-16 16:30:19.178578+00 | 2024-10-16 16:30:19.26324+00 | 2024-10-16 16:30:19.263267+00 | Client | ClientRead | RELEASE SAVEPOINT cockroach_restart 00:00:00.084768 | 33139 | zitadel_es_pusher | 2024-10-16 16:29:54.979585+00 | 2024-10-16 16:30:19.180762+00 | 2024-10-16 16:30:19.26553+00 | 2024-10-16 16:30:19.265531+00 | LWLock | WALWriteLock | commit 00:00:00.077377 | 33136 | zitadel_es_pusher | 2024-10-16 16:29:54.978582+00 | 2024-10-16 16:30:19.187883+00 | 2024-10-16 16:30:19.26526+00 | 2024-10-16 16:30:19.265431+00 | Client | ClientRead | WITH existing AS ( + | | | | | | | | | (SELECT instance_id, aggregate_type, aggregate_id, "sequence" FROM eventstore.events2 WHERE instance_id = $1 AND aggregate_type = $2 AND aggregate_id = $3 ORDER BY "sequence" DE 00:00:00.012309 | 33123 | zitadel_es_pusher | 2024-10-16 16:29:54.963484+00 | 2024-10-16 16:30:19.175066+00 | 2024-10-16 16:30:19.187375+00 | 2024-10-16 16:30:19.187376+00 | IO | WalSync | commit 00:00:00 | 33034 | zitadel_queries | 2024-10-16 16:25:53.90791+00 | 2024-10-16 16:30:19.262921+00 | 2024-10-16 16:30:19.262921+00 | 2024-10-16 16:30:19.263133+00 | Client | ClientRead | begin isolation level read committed read only 00:00:00 | 33039 | zitadel_queries | 2024-10-16 16:25:53.914106+00 | 2024-10-16 16:30:19.191676+00 | 2024-10-16 16:30:19.191676+00 | 2024-10-16 16:30:19.191687+00 | Client | ClientRead | begin isolation level read committed read only 00:00:00.24539 | 33083 | zitadel_projection_spooler | 2024-10-16 16:27:49.895548+00 | 2024-10-16 16:30:19.020058+00 | 2024-10-16 16:30:19.265448+00 | 2024-10-16 16:30:19.26546+00 | Client | ClientRead | SAVEPOINT exec_stmt 00:00:00 | 33125 | zitadel_es_pusher | 2024-10-16 16:29:54.963859+00 | 2024-10-16 16:30:19.191715+00 | 2024-10-16 16:30:19.191715+00 | 2024-10-16 16:30:19.191729+00 | Client | ClientRead | begin 00:00:00.004292 | 33032 | zitadel_queries | 2024-10-16 16:25:53.906624+00 | 2024-10-16 16:30:19.187713+00 | 2024-10-16 16:30:19.192005+00 | 2024-10-16 16:30:19.192062+00 | Client | ClientRead | SELECT created_at, event_type, "sequence", "position", payload, creator, "owner", instance_id, aggregate_type, aggregate_id, revision FROM eventstore.events2 WHERE instance_id = $1 AND aggregate_type 00:00:00 | 33031 | zitadel_queries | 2024-10-16 16:25:53.906422+00 | 2024-10-16 16:30:19.191625+00 | 2024-10-16 16:30:19.191625+00 | 2024-10-16 16:30:19.191645+00 | Client | ClientRead | begin isolation level read committed read only ``` The amount of idle transactions is significantly less if the query transactions are removed: example: ``` ?column? | pid | application_name | backend_start | xact_start | query_start | state_change | wait_event_type | wait_event | query -----------------+-------+----------------------------+-------------------------------+-------------------------------+-------------------------------+-------------------------------+-----------------+------------+---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- 00:00:00.000094 | 32741 | zitadel_queries | 2024-10-16 16:23:49.73935+00 | 2024-10-16 16:24:59.785589+00 | 2024-10-16 16:24:59.785683+00 | 2024-10-16 16:24:59.785684+00 | | | SELECT created_at, event_type, "sequence", "position", payload, creator, "owner", instance_id, aggregate_type, aggregate_id, revision FROM eventstore.events2 WHERE instance_id = $1 AND aggregate_type 00:00:00 | 32762 | zitadel_es_pusher | 2024-10-16 16:24:02.275136+00 | 2024-10-16 16:24:59.784586+00 | 2024-10-16 16:24:59.784586+00 | 2024-10-16 16:24:59.784607+00 | Client | ClientRead | begin 00:00:00.000167 | 32742 | zitadel_queries | 2024-10-16 16:23:49.740489+00 | 2024-10-16 16:24:59.784274+00 | 2024-10-16 16:24:59.784441+00 | 2024-10-16 16:24:59.784442+00 | | | with usr as ( + | | | | | | | | | select u.id, u.creation_date, u.change_date, u.sequence, u.state, u.resource_owner, u.username, n.login_name as preferred_login_name + | | | | | | | | | from projections.users13 u + | | | | | | | | | left join projections.l 00:00:00.256014 | 32759 | zitadel_projection_spooler | 2024-10-16 16:24:01.418429+00 | 2024-10-16 16:24:59.52959+00 | 2024-10-16 16:24:59.785604+00 | 2024-10-16 16:24:59.785649+00 | Client | ClientRead | UPDATE projections.milestones SET reached_date = $1 WHERE (instance_id = $2) AND (type = $3) AND (reached_date IS NULL) 00:00:00.014199 | 32773 | zitadel_es_pusher | 2024-10-16 16:24:02.320404+00 | 2024-10-16 16:24:59.769509+00 | 2024-10-16 16:24:59.783708+00 | 2024-10-16 16:24:59.783709+00 | IO | WalSync | commit 00:00:00 | 32765 | zitadel_es_pusher | 2024-10-16 16:24:02.28173+00 | 2024-10-16 16:24:59.780413+00 | 2024-10-16 16:24:59.780413+00 | 2024-10-16 16:24:59.780426+00 | Client | ClientRead | begin 00:00:00.012729 | 32777 | zitadel_es_pusher | 2024-10-16 16:24:02.339737+00 | 2024-10-16 16:24:59.767432+00 | 2024-10-16 16:24:59.780161+00 | 2024-10-16 16:24:59.780195+00 | Client | ClientRead | RELEASE SAVEPOINT cockroach_restart ``` --------- Co-authored-by: Tim Möhlmann <tim+github@zitadel.com> Co-authored-by: Livio Spring <livio.a@gmail.com> Co-authored-by: Max Peintner <max@caos.ch> Co-authored-by: Elio Bischof <elio@zitadel.com> Co-authored-by: Stefan Benz <46600784+stebenz@users.noreply.github.com> Co-authored-by: Miguel Cabrerizo <30386061+doncicuto@users.noreply.github.com> Co-authored-by: Joakim Lodén <Loddan@users.noreply.github.com> Co-authored-by: Yxnt <Yxnt@users.noreply.github.com> Co-authored-by: Stefan Benz <stefan@caos.ch> Co-authored-by: Harsha Reddy <harsha.reddy@klaviyo.com> Co-authored-by: Zach H <zhirschtritt@gmail.com>
188 lines
4.8 KiB
Go
188 lines
4.8 KiB
Go
package eventstore
|
|
|
|
import (
|
|
"context"
|
|
"database/sql"
|
|
_ "embed"
|
|
"errors"
|
|
"fmt"
|
|
"strconv"
|
|
"strings"
|
|
|
|
"github.com/cockroachdb/cockroach-go/v2/crdb"
|
|
"github.com/jackc/pgx/v5/pgconn"
|
|
"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, span := tracing.NewSpan(ctx)
|
|
defer func() { span.EndWithError(err) }()
|
|
|
|
tx, err := es.client.BeginTx(ctx, nil)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
// tx is not closed because [crdb.ExecuteInTx] takes care of that
|
|
var (
|
|
sequences []*latestSequence
|
|
)
|
|
|
|
err = crdb.ExecuteInTx(ctx, &transaction{tx}, func() (err error) {
|
|
inTxCtx, span := tracing.NewSpan(ctx)
|
|
defer func() { span.EndWithError(err) }()
|
|
|
|
sequences, err = latestSequences(inTxCtx, tx, commands)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
events, err = insertEvents(inTxCtx, tx, sequences, commands)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
|
|
if err = handleUniqueConstraints(inTxCtx, tx, commands); err != nil {
|
|
return err
|
|
}
|
|
|
|
// CockroachDB by default does not allow multiple modifications of the same table using ON CONFLICT
|
|
// Thats why we enable it manually
|
|
if es.client.Type() == "cockroach" {
|
|
_, err = tx.Exec("SET enable_multiple_modifications_of_table = on")
|
|
if err != nil {
|
|
return err
|
|
}
|
|
}
|
|
|
|
return handleFieldCommands(inTxCtx, tx, commands)
|
|
})
|
|
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
return events, nil
|
|
}
|
|
|
|
//go:embed push.sql
|
|
var pushStmt string
|
|
|
|
func insertEvents(ctx context.Context, tx *sql.Tx, sequences []*latestSequence, commands []eventstore.Command) ([]eventstore.Event, error) {
|
|
events, placeholders, args, err := mapCommands(commands, sequences)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
|
|
rows, err := tx.QueryContext(ctx, fmt.Sprintf(pushStmt, strings.Join(placeholders, ", ")), args...)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
|
|
for i := 0; rows.Next(); i++ {
|
|
err = rows.Scan(&events[i].(*event).createdAt, &events[i].(*event).position)
|
|
if err != nil {
|
|
logging.WithError(err).Warn("failed to scan events")
|
|
return nil, err
|
|
}
|
|
}
|
|
|
|
if err := rows.Err(); err != nil {
|
|
pgErr := new(pgconn.PgError)
|
|
if errors.As(err, &pgErr) {
|
|
// Check if push tries to write an event just written
|
|
// by another transaction
|
|
if pgErr.Code == "40001" {
|
|
// TODO: @livio-a should we return the parent or not?
|
|
return nil, zerrors.ThrowInvalidArgument(err, "V3-p5xAn", "Errors.AlreadyExists")
|
|
}
|
|
}
|
|
logging.WithError(rows.Err()).Warn("failed to push events")
|
|
return nil, zerrors.ThrowInternal(err, "V3-VGnZY", "Errors.Internal")
|
|
}
|
|
|
|
return events, nil
|
|
}
|
|
|
|
const argsPerCommand = 10
|
|
|
|
func mapCommands(commands []eventstore.Command, sequences []*latestSequence) (events []eventstore.Event, placeholders []string, args []any, err error) {
|
|
events = make([]eventstore.Event, len(commands))
|
|
args = make([]any, 0, len(commands)*argsPerCommand)
|
|
placeholders = make([]string, len(commands))
|
|
|
|
for i, command := range commands {
|
|
sequence := searchSequenceByCommand(sequences, command)
|
|
if sequence == nil {
|
|
logging.WithFields(
|
|
"aggType", command.Aggregate().Type,
|
|
"aggID", command.Aggregate().ID,
|
|
"instance", command.Aggregate().InstanceID,
|
|
).Panic("no sequence found")
|
|
// added return for linting
|
|
return nil, nil, nil, nil
|
|
}
|
|
sequence.sequence++
|
|
|
|
events[i], err = commandToEvent(sequence, command)
|
|
if err != nil {
|
|
return nil, nil, nil, err
|
|
}
|
|
|
|
placeholders[i] = fmt.Sprintf(pushPlaceholderFmt,
|
|
i*argsPerCommand+1,
|
|
i*argsPerCommand+2,
|
|
i*argsPerCommand+3,
|
|
i*argsPerCommand+4,
|
|
i*argsPerCommand+5,
|
|
i*argsPerCommand+6,
|
|
i*argsPerCommand+7,
|
|
i*argsPerCommand+8,
|
|
i*argsPerCommand+9,
|
|
i*argsPerCommand+10,
|
|
)
|
|
|
|
revision, err := strconv.Atoi(strings.TrimPrefix(string(events[i].(*event).aggregate.Version), "v"))
|
|
if err != nil {
|
|
return nil, nil, nil, zerrors.ThrowInternal(err, "V3-JoZEp", "Errors.Internal")
|
|
}
|
|
args = append(args,
|
|
events[i].(*event).aggregate.InstanceID,
|
|
events[i].(*event).aggregate.ResourceOwner,
|
|
events[i].(*event).aggregate.Type,
|
|
events[i].(*event).aggregate.ID,
|
|
revision,
|
|
events[i].(*event).creator,
|
|
events[i].(*event).typ,
|
|
events[i].(*event).payload,
|
|
events[i].(*event).sequence,
|
|
i,
|
|
)
|
|
}
|
|
|
|
return events, placeholders, args, nil
|
|
}
|
|
|
|
type transaction struct {
|
|
*sql.Tx
|
|
}
|
|
|
|
var _ crdb.Tx = (*transaction)(nil)
|
|
|
|
func (t *transaction) Exec(ctx context.Context, query string, args ...interface{}) error {
|
|
_, err := t.Tx.ExecContext(ctx, query, args...)
|
|
return err
|
|
}
|
|
|
|
func (t *transaction) Commit(ctx context.Context) error {
|
|
return t.Tx.Commit()
|
|
}
|
|
|
|
func (t *transaction) Rollback(ctx context.Context) error {
|
|
return t.Tx.Rollback()
|
|
}
|