zitadel/cmd/initialise/verify_zitadel.go
Silvan dab5d9e756
refactor(eventstore): move push logic to sql (#8816)
# Which Problems Are Solved

If many events are written to the same aggregate id it can happen that
zitadel [starts to retry the push
transaction](48ffc902cc/internal/eventstore/eventstore.go (L101))
because [the locking
behaviour](48ffc902cc/internal/eventstore/v3/sequence.go (L25))
during push does compute the wrong sequence because newly committed
events are not visible to the transaction. These events impact the
current sequence.

In cases with high command traffic on a single aggregate id this can
have severe impact on general performance of zitadel. Because many
connections of the `eventstore pusher` database pool are blocked by each
other.

# How the Problems Are Solved

To improve the performance this locking mechanism was removed and the
business logic of push is moved to sql functions which reduce network
traffic and can be analyzed by the database before the actual push. For
clients of the eventstore framework nothing changed.

# Additional Changes

- after a connection is established prefetches the newly added database
types
- `eventstore.BaseEvent` now returns the correct revision of the event

# Additional Context

- part of https://github.com/zitadel/zitadel/issues/8931

---------

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>
2024-12-04 13:51:40 +00:00

143 lines
3.4 KiB
Go

package initialise
import (
"context"
"database/sql"
_ "embed"
"fmt"
"github.com/spf13/cobra"
"github.com/spf13/viper"
"github.com/zitadel/logging"
"github.com/zitadel/zitadel/internal/database"
"github.com/zitadel/zitadel/internal/database/dialect"
es_v3 "github.com/zitadel/zitadel/internal/eventstore/v3"
)
func newZitadel() *cobra.Command {
return &cobra.Command{
Use: "zitadel",
Short: "initialize ZITADEL internals",
Long: `initialize ZITADEL internals.
Prerequisites:
- cockroachDB or postgreSQL with user and database
`,
Run: func(cmd *cobra.Command, args []string) {
config := MustNewConfig(viper.GetViper())
err := verifyZitadel(cmd.Context(), config.Database)
logging.OnError(err).Fatal("unable to init zitadel")
},
}
}
func VerifyZitadel(ctx context.Context, db *database.DB, config database.Config) error {
err := ReadStmts(config.Type())
if err != nil {
return err
}
conn, err := db.Conn(ctx)
if err != nil {
return err
}
defer conn.Close()
logging.WithFields().Info("verify system")
if err := exec(ctx, conn, fmt.Sprintf(createSystemStmt, config.Username()), nil); err != nil {
return err
}
logging.WithFields().Info("verify encryption keys")
if err := createEncryptionKeys(ctx, conn); err != nil {
return err
}
logging.WithFields().Info("verify projections")
if err := exec(ctx, conn, fmt.Sprintf(createProjectionsStmt, config.Username()), nil); err != nil {
return err
}
logging.WithFields().Info("verify eventstore")
if err := exec(ctx, conn, fmt.Sprintf(createEventstoreStmt, config.Username()), nil); err != nil {
return err
}
logging.WithFields().Info("verify events tables")
if err := createEvents(ctx, conn); err != nil {
return err
}
logging.WithFields().Info("verify system sequence")
if err := exec(ctx, conn, createSystemSequenceStmt, nil); err != nil {
return err
}
logging.WithFields().Info("verify unique constraints")
if err := exec(ctx, conn, createUniqueConstraints, nil); err != nil {
return err
}
return nil
}
func verifyZitadel(ctx context.Context, config database.Config) error {
logging.WithFields("database", config.DatabaseName()).Info("verify zitadel")
db, err := database.Connect(config, false, dialect.DBPurposeQuery)
if err != nil {
return err
}
if err := VerifyZitadel(ctx, db, config); err != nil {
return err
}
return db.Close()
}
func createEncryptionKeys(ctx context.Context, db database.Beginner) error {
tx, err := db.BeginTx(ctx, nil)
if err != nil {
return err
}
if _, err = tx.Exec(createEncryptionKeysStmt); err != nil {
rollbackErr := tx.Rollback()
logging.OnError(rollbackErr).Error("rollback failed")
return err
}
return tx.Commit()
}
func createEvents(ctx context.Context, conn *sql.Conn) (err error) {
tx, err := conn.BeginTx(ctx, nil)
if err != nil {
return err
}
defer func() {
if err != nil {
rollbackErr := tx.Rollback()
logging.OnError(rollbackErr).Error("rollback failed")
return
}
err = tx.Commit()
}()
// if events already exists events2 is created during a setup job
var count int
row := tx.QueryRow("SELECT count(*) FROM information_schema.tables WHERE table_schema = 'eventstore' AND table_name like 'events%'")
if err = row.Scan(&count); err != nil {
return err
}
if row.Err() != nil || count >= 1 {
return row.Err()
}
_, err = tx.Exec(createEventsStmt)
if err != nil {
return err
}
return es_v3.CheckExecutionPlan(ctx, conn)
}